Implements durabletask ListInstanceIDs & GetInstanceHistory

Today, there is no ways of discovering the list of workflow instances that are currently running or have completed in the past without using external storage queries.
The Dapr CLI [introduced list and workflow history commands](dapr/cli#1560) to get information about running and completed workflows, however these commands rely on direct queries to the underlying storage provider.
By introducing this functionality into the durabletask framework itself, these commands need only talk to Daprd, removing the requirement for direct access to the storage provider as well as authentication.
Daprd can make these queries itself, and use the Actor State Store component to access the underlying storage.

Implements the new durabletask APIs according to dapr/proposals#93

Update components-contrib & durabletask-go to origin

Use shared const for ActorTypePrefix

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/cloudevents/sdk-go/v2 v2.15.2
 	github.com/coreos/go-oidc/v3 v3.14.1
-	github.com/dapr/components-contrib v1.16.1
+	github.com/dapr/components-contrib v1.16.2-0.20251113171451-b78f056c8491
 	github.com/dapr/durabletask-go v0.10.2-0.20251113171253-87ecdf8f0547
 	github.com/dapr/kit v0.16.2-0.20251117143824-2fd5d0c93524
 	github.com/diagridio/go-etcd-cron v0.10.0

go.sum

@@ -515,8 +515,8 @@ github.com/dancannon/gorethink v4.0.0+incompatible h1:KFV7Gha3AuqT+gr0B/eKvGhbjm
 github.com/dancannon/gorethink v4.0.0+incompatible/go.mod h1:BLvkat9KmZc1efyYwhz3WnybhRZtgF1K929FD8z1avU=
 github.com/danieljoos/wincred v1.1.2 h1:QLdCxFs1/Yl4zduvBdcHB8goaYk9RARS2SgLLRuAyr0=
 github.com/danieljoos/wincred v1.1.2/go.mod h1:GijpziifJoIBfYh+S7BbkdUTU4LfM+QnGqR5Vl2tAx0=
-github.com/dapr/components-contrib v1.16.1 h1:EpntHk5qUXTbB55kec97cMQbC2QXBwbBU6QMI3ljV8g=
-github.com/dapr/components-contrib v1.16.1/go.mod h1:1AufCWqZwBj//UkyS7FesOEmp5/E6Xgy1tyCn8peiR4=
+github.com/dapr/components-contrib v1.16.2-0.20251113171451-b78f056c8491 h1:ms0IhiGK6Mow+e1DZgIxD3qo7xqOPjpxeFkhahBo+Tg=
+github.com/dapr/components-contrib v1.16.2-0.20251113171451-b78f056c8491/go.mod h1:1AufCWqZwBj//UkyS7FesOEmp5/E6Xgy1tyCn8peiR4=
 github.com/dapr/durabletask-go v0.10.2-0.20251113171253-87ecdf8f0547 h1:bD4JBlXDHURsgvhIB1HQ1q0k8kYfVo/iNSBi0guSoe0=
 github.com/dapr/durabletask-go v0.10.2-0.20251113171253-87ecdf8f0547/go.mod h1:0Ts4rXp74JyG19gDWPcwNo5V6NBZzhARzHF5XynmA7Q=
 github.com/dapr/kit v0.16.2-0.20251117143824-2fd5d0c93524 h1:SQ7VeWGnypENpGjsL94wN2IgH+oYHx9ULpYrpFoRExQ=

pkg/runtime/runtime.go

@@ -307,6 +307,7 @@ func newDaprRuntime(ctx context.Context,
 		Resiliency:                resiliencyProvider,
 		EventSink:                 runtimeConfig.workflowEventSink,
 		EnableClusteredDeployment: globalConfig.IsFeatureEnabled(config.WorkflowsClusteredDeployment),
+		ComponentStore:            compStore,
 	})
 
 	jobsManager, err := scheduler.New(scheduler.Options{

pkg/runtime/wfengine/backends/actors/actors.go

@@ -43,7 +43,9 @@ import (
 	invokev1 "github.com/dapr/dapr/pkg/messaging/v1"
 	internalsv1pb "github.com/dapr/dapr/pkg/proto/internals/v1"
 	"github.com/dapr/dapr/pkg/resiliency"
+	"github.com/dapr/dapr/pkg/runtime/compstore"
 	"github.com/dapr/dapr/pkg/runtime/wfengine/state"
+	"github.com/dapr/dapr/pkg/runtime/wfengine/state/list"
 	"github.com/dapr/dapr/pkg/runtime/wfengine/todo"
 	"github.com/dapr/dapr/utils"
 	"github.com/dapr/durabletask-go/api"
@@ -62,15 +64,15 @@ const (
 	WorkflowNameLabelKey = "workflow"
 	ActivityNameLabelKey = "activity"
 	ExecutorNameLabelKey = "executor"
-	ActorTypePrefix      = "dapr.internal."
 )
 
 type Options struct {
-	AppID      string
-	Namespace  string
-	Actors     actors.Interface
-	Resiliency resiliency.Provider
-	EventSink  orchestrator.EventSink
+	AppID          string
+	Namespace      string
+	Actors         actors.Interface
+	Resiliency     resiliency.Provider
+	EventSink      orchestrator.EventSink
+	ComponentStore *compstore.ComponentStore
 	// experimental feature
 	// enabling this will use the cluster tasks backend for pending tasks, instead of the default local implementation
 	// the cluster tasks backend uses actors to share the state of pending tasks
@@ -90,6 +92,7 @@ type Actors struct {
 	resiliency                resiliency.Provider
 	actors                    actors.Interface
 	eventSink                 orchestrator.EventSink
+	compStore                 *compstore.ComponentStore
 
 	orchestrationWorkItemChan chan *backend.OrchestrationWorkItem
 	activityWorkItemChan      chan *backend.ActivityWorkItem
@@ -102,19 +105,20 @@ func New(opts Options) *Actors {
 	if opts.EnableClusteredDeployment {
 		pendingTasksBackend = NewClusterTasksBackend(ClusterTasksBackendOptions{
 			Actors:            opts.Actors,
-			ExecutorActorType: ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + ExecutorNameLabelKey,
+			ExecutorActorType: todo.ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + ExecutorNameLabelKey,
 		})
 	}
 	return &Actors{
 		appID:                     opts.AppID,
 		namespace:                 opts.Namespace,
-		workflowActorType:         ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + WorkflowNameLabelKey,
-		activityActorType:         ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + ActivityNameLabelKey,
-		executorActorType:         ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + ExecutorNameLabelKey,
+		workflowActorType:         todo.ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + WorkflowNameLabelKey,
+		activityActorType:         todo.ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + ActivityNameLabelKey,
+		executorActorType:         todo.ActorTypePrefix + opts.Namespace + utils.DotDelimiter + opts.AppID + utils.DotDelimiter + ExecutorNameLabelKey,
 		actors:                    opts.Actors,
 		resiliency:                opts.Resiliency,
 		pendingTasksBackend:       pendingTasksBackend,
 		enableClusteredDeployment: opts.EnableClusteredDeployment,
+		compStore:                 opts.ComponentStore,
 		orchestrationWorkItemChan: make(chan *backend.OrchestrationWorkItem, 1),
 		activityWorkItemChan:      make(chan *backend.ActivityWorkItem, 1),
 		eventSink:                 opts.EventSink,
@@ -630,6 +634,42 @@ func (abe *Actors) WaitForOrchestratorCompletion(ctx context.Context, request *p
 	return abe.pendingTasksBackend.WaitForOrchestratorCompletion(ctx, request)
 }
 
+func (abe *Actors) ListInstanceIDs(ctx context.Context, req *protos.ListInstanceIDsRequest) (*protos.ListInstanceIDsResponse, error) {
+	resp, err := list.ListInstanceIDs(ctx, list.ListOptions{
+		ComponentStore:    abe.compStore,
+		Namespace:         abe.namespace,
+		AppID:             abe.appID,
+		PageSize:          req.PageSize,          //nolint:protogetter
+		ContinuationToken: req.ContinuationToken, //nolint:protogetter
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &protos.ListInstanceIDsResponse{
+		InstanceIds:       resp.Keys,
+		ContinuationToken: resp.ContinuationToken,
+	}, nil
+}
+
+func (abe *Actors) GetInstanceHistory(ctx context.Context, req *protos.GetInstanceHistoryRequest) (*protos.GetInstanceHistoryResponse, error) {
+	ss, err := abe.actors.State(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := state.LoadWorkflowState(ctx, ss, req.GetInstanceId(), state.Options{
+		AppID:             abe.appID,
+		WorkflowActorType: abe.workflowActorType,
+		ActivityActorType: abe.activityActorType,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &protos.GetInstanceHistoryResponse{Events: resp.History}, nil
+}
+
 func (abe *Actors) purgeWorkflow(ctx context.Context, id api.InstanceID) error {
 	req := internalsv1pb.
 		NewInternalInvokeRequest(todo.PurgeWorkflowStateMethod).
@@ -702,11 +742,3 @@ func (abe *Actors) purgeWorkflowForce(ctx context.Context, id api.InstanceID) er
 		},
 	)
 }
-
-func (abe *Actors) GetInstanceHistory(ctx context.Context, req *protos.GetInstanceHistoryRequest) (*protos.GetInstanceHistoryResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "GetInstanceHistory is not implemented in the Actors backend")
-}
-
-func (abe *Actors) ListInstanceIDs(ctx context.Context, req *protos.ListInstanceIDsRequest) (*protos.ListInstanceIDsResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "ListInstanceIDs is not implemented in the Actors backend")
-}

pkg/runtime/wfengine/state/list/list.go

@@ -0,0 +1,77 @@
+/*
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package list
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/dapr/components-contrib/state"
+	"github.com/dapr/dapr/pkg/runtime/compstore"
+	"github.com/dapr/dapr/pkg/runtime/wfengine/todo"
+)
+
+type ListOptions struct {
+	ComponentStore *compstore.ComponentStore
+	Namespace      string
+	AppID          string
+
+	PageSize          *uint32
+	ContinuationToken *string
+}
+
+type ListInstanceIDsResult struct {
+	Keys              []string
+	ContinuationToken *string
+}
+
+func ListInstanceIDs(ctx context.Context, opts ListOptions) (*ListInstanceIDsResult, error) {
+	store, _, ok := opts.ComponentStore.GetStateStoreActor()
+	if !ok {
+		return nil, errors.New("no state store with actor support found")
+	}
+
+	ks, ok := store.(state.KeysLiker)
+	if !ok {
+		return nil, fmt.Errorf("state store %T does not support listing keys", store)
+	}
+
+	like := opts.AppID + "||" + todo.ActorTypePrefix + opts.Namespace + "." + opts.AppID + ".workflow||%||metadata"
+
+	resp, err := ks.KeysLike(ctx, &state.KeysLikeRequest{
+		Pattern:           like,
+		ContinuationToken: opts.ContinuationToken,
+		PageSize:          opts.PageSize,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	keys := make([]string, 0, len(resp.Keys))
+	for _, key := range resp.Keys {
+		split := strings.Split(key, "||")
+		if len(split) != 4 {
+			return nil, fmt.Errorf("invalid key format: %s", key)
+		}
+
+		keys = append(keys, split[2])
+	}
+
+	return &ListInstanceIDsResult{
+		Keys:              keys,
+		ContinuationToken: resp.ContinuationToken,
+	}, nil
+}

pkg/runtime/wfengine/todo/todo.go

@@ -32,6 +32,8 @@ const (
 	RerunWorkflowInstance        = "RerunWorkflowInstance"
 
 	MetadataActivityReminderDueTime = "dueTime"
+
+	ActorTypePrefix = "dapr.internal."
 )
 
 var (

pkg/runtime/wfengine/wfengine.go

@@ -30,6 +30,7 @@ import (
 	"github.com/dapr/dapr/pkg/config"
 	runtimev1pb "github.com/dapr/dapr/pkg/proto/runtime/v1"
 	"github.com/dapr/dapr/pkg/resiliency"
+	"github.com/dapr/dapr/pkg/runtime/compstore"
 	"github.com/dapr/dapr/pkg/runtime/processor"
 	backendactors "github.com/dapr/dapr/pkg/runtime/wfengine/backends/actors"
 	"github.com/dapr/durabletask-go/backend"
@@ -59,6 +60,7 @@ type Options struct {
 	Resiliency                resiliency.Provider
 	EventSink                 orchestrator.EventSink
 	EnableClusteredDeployment bool
+	ComponentStore            *compstore.ComponentStore
 }
 
 type engine struct {
@@ -83,6 +85,7 @@ func New(opts Options) Interface {
 		Resiliency:                opts.Resiliency,
 		EventSink:                 opts.EventSink,
 		EnableClusteredDeployment: opts.EnableClusteredDeployment,
+		ComponentStore:            opts.ComponentStore,
 	})
 
 	var getWorkItemsCount atomic.Int32

tests/integration/suite/daprd/hotreload/operator/actorstate.go

@@ -176,7 +176,7 @@ func (a *actorstate) Run(t *testing.T, ctx context.Context) {
 	require.ElementsMatch(t, []*rtv1.RegisteredComponents{
 		{
 			Name: "mystore", Type: "state.in-memory", Version: "v1",
-			Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "ACTOR"},
+			Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "KEYS_LIKE", "ACTOR"},
 		},
 	}, comps)
 	inmemStore.Spec.Metadata = []common.NameValuePair{}
@@ -188,7 +188,7 @@ func (a *actorstate) Run(t *testing.T, ctx context.Context) {
 	require.ElementsMatch(t, []*rtv1.RegisteredComponents{
 		{
 			Name: "mystore", Type: "state.in-memory", Version: "v1",
-			Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "ACTOR"},
+			Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "KEYS_LIKE", "ACTOR"},
 		},
 	}, comps)
 	a.operatorDelete.SetComponents()

tests/integration/suite/daprd/hotreload/operator/crypto.go

@@ -217,7 +217,7 @@ func (c *crypto) Run(t *testing.T, ctx context.Context) {
 				{Name: "crypto3", Type: "crypto.dapr.localstorage", Version: "v1"},
 				{
 					Name: "crypto2", Type: "state.in-memory", Version: "v1",
-					Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "ACTOR"},
+					Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "KEYS_LIKE", "ACTOR"},
 				},
 			}, resp.GetRegisteredComponents())
 		}, time.Second*10, time.Millisecond*10)
@@ -255,7 +255,7 @@ func (c *crypto) Run(t *testing.T, ctx context.Context) {
 			assert.ElementsMatch(c, []*rtv1.RegisteredComponents{
 				{
 					Name: "crypto3", Type: "state.in-memory", Version: "v1",
-					Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "ACTOR"},
+					Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "KEYS_LIKE", "ACTOR"},
 				},
 			}, resp.GetRegisteredComponents())
 		}, time.Second*10, time.Millisecond*10)

tests/integration/suite/daprd/hotreload/operator/informer/components.go

@@ -141,7 +141,7 @@ func (c *components) Run(t *testing.T, ctx context.Context) {
 		exp := []*rtv1.RegisteredComponents{
 			{
 				Name: "123", Type: "state.in-memory", Version: "v1",
-				Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "ACTOR"},
+				Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "DELETE_WITH_PREFIX", "KEYS_LIKE", "ACTOR"},
 			},
 		}
 		require.EventuallyWithT(t, func(ct *assert.CollectT) {
@@ -179,7 +179,7 @@ func (c *components) Run(t *testing.T, ctx context.Context) {
 			exp := []*rtv1.RegisteredComponents{
 				{
 					Name: "123", Type: "state.sqlite", Version: "v1",
-					Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "ACTOR"},
+					Capabilities: []string{"ETAG", "TRANSACTIONAL", "TTL", "KEYS_LIKE", "ACTOR"},
 				},
 			}
 			assert.ElementsMatch(ct, exp, c.daprd1.GetMetaRegisteredComponents(ct, ctx))