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

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

Author: JoshVanL

go.mod

@@ -525,3 +525,7 @@ replace (
 //
 // Then, run `make modtidy-all` in this repository.
 // This ensures that go.mod and go.sum are up-to-date for each go.mod file.
+
+replace github.com/dapr/durabletask-go => github.com/joshvanl/durabletask-go v0.0.0-20251105162850-0096c0be3cae
+
+replace github.com/dapr/components-contrib => github.com/joshvanl/components-contrib v0.0.0-20251105114938-38d2fb353cc5

go.sum

@@ -515,10 +515,6 @@ 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/durabletask-go v0.10.1 h1:gE88Qh4+/6zKdegHjOAOx+UQaPxmwWKWoIDivee23XY=
-github.com/dapr/durabletask-go v0.10.1/go.mod h1:0Ts4rXp74JyG19gDWPcwNo5V6NBZzhARzHF5XynmA7Q=
 github.com/dapr/kit v0.16.1 h1:MqLAhHVg8trPy2WJChMZFU7ToeondvxcNHYVvMDiVf4=
 github.com/dapr/kit v0.16.1/go.mod h1:40ZWs5P6xfYf7O59XgwqZkIyDldTIXlhTQhGop8QoSM=
 github.com/dave/jennifer v1.4.0/go.mod h1:fIb+770HOpJ2fmN9EPPKOqm1vMGhB+TwXKMZhrIygKg=
@@ -1132,6 +1128,10 @@ github.com/jonboulle/clockwork v0.5.0 h1:Hyh9A8u51kptdkR+cqRpT1EebBwTn1oK9YfGYbd
 github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7XN3SzBPjZF60=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
+github.com/joshvanl/components-contrib v0.0.0-20251105114938-38d2fb353cc5 h1:IgQW5uO5mvT2jTfpAeVecJ9aTSuRPS/y9g1SU8XmPn0=
+github.com/joshvanl/components-contrib v0.0.0-20251105114938-38d2fb353cc5/go.mod h1:1AufCWqZwBj//UkyS7FesOEmp5/E6Xgy1tyCn8peiR4=
+github.com/joshvanl/durabletask-go v0.0.0-20251105162850-0096c0be3cae h1:cWA+L6X+Kq3CzzwB3U4kQ8nmPs7e6UwhAEnTIh8hSOo=
+github.com/joshvanl/durabletask-go v0.0.0-20251105162850-0096c0be3cae/go.mod h1:0Ts4rXp74JyG19gDWPcwNo5V6NBZzhARzHF5XynmA7Q=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.5/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=

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

@@ -42,7 +42,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"
@@ -64,11 +66,12 @@ const (
 )
 
 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
@@ -88,6 +91,7 @@ type Actors struct {
 	resiliency                resiliency.Provider
 	actors                    actors.Interface
 	eventSink                 orchestrator.EventSink
+	compStore                 *compstore.ComponentStore
 
 	orchestrationWorkItemChan chan *backend.OrchestrationWorkItem
 	activityWorkItemChan      chan *backend.ActivityWorkItem
@@ -113,6 +117,7 @@ func New(opts Options) *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,
@@ -629,3 +634,39 @@ func (abe *Actors) WaitForActivityCompletion(ctx context.Context, request *proto
 func (abe *Actors) WaitForOrchestratorCompletion(ctx context.Context, request *protos.OrchestratorRequest) (*protos.OrchestratorResponse, error) {
 	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,
+		ContinuationToken: req.ContinuationToken,
+	})
+	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
+}

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

@@ -0,0 +1,76 @@
+/*
+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"
+)
+
+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 + "||dapr.internal." + opts.Namespace + "." + opts.AppID + ".workflow||%||metadata"
+
+	resp, err := ks.KeysLike(ctx, &state.KeysLikeRequest{
+		Pattern:       like,
+		ContinueToken: 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.ContinueToken,
+	}, nil
+}

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/framework/process/workflow/workflow.go

@@ -30,6 +30,7 @@ import (
 	"github.com/dapr/dapr/tests/integration/framework/process/sqlite"
 	"github.com/dapr/durabletask-go/client"
 	"github.com/dapr/durabletask-go/task"
+	"github.com/dapr/durabletask-go/workflow"
 )
 
 type Workflow struct {
@@ -83,6 +84,11 @@ func New(t *testing.T, fopts ...Option) *Workflow {
 			}
 		}
 
+		// TODO: @joshvanl
+		//if i > 0 {
+		//	dopts = append(dopts, daprd.WithAppID(daprds[0].AppID()))
+		//}
+
 		daprds[i] = daprd.New(t, dopts...)
 	}
 
@@ -165,6 +171,17 @@ func (w *Workflow) BackendClient(t *testing.T, ctx context.Context) *client.Task
 	return w.BackendClientN(t, ctx, 0)
 }
 
+func (w *Workflow) WorkflowClient(t *testing.T, ctx context.Context) *workflow.Client {
+	t.Helper()
+	return workflow.NewClient(w.Dapr().GRPCConn(t, ctx))
+}
+
+func (w *Workflow) WorkflowClientN(t *testing.T, ctx context.Context, index int) *workflow.Client {
+	t.Helper()
+	require.Less(t, index, len(w.daprds), "index out of range")
+	return workflow.NewClient(w.DaprN(index).GRPCConn(t, ctx))
+}
+
 // BackendClient returns a backend client for the specified index
 func (w *Workflow) BackendClientN(t *testing.T, ctx context.Context, index int) *client.TaskHubGrpcClient {
 	t.Helper()

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))