Workflow: Rerun workflow instance form Event ID

Change extends the functionally with an rpc to enable rerunning a
workflow from an existing _terminal_ workflow instance. See;
dapr/proposals#80

Rerunning workflows has the given restrictions;
- The source workflow instance _must_ be in a terminal state- i.e. is
  not running or suspended. It has to have been run before.
- The target event ID must be an activity. The event ID must exist.
- The target event ID input data can be overwritten.
- The new instance ID can be supplied, or will be randomly generated as
  a UUID.
- The source and target instance ID cannot be the same. The orginal
  history will not be overwritten/deleted/truncated.

The new workflow cannot have any timers or raise events which are active
at the time of starting. There is no technical resource why we cannot
achieve this, but is a limitation of the existing implementation. A
future change should see a refactor to enable this; likely moving timers
and raise events to a new actor type and instantiation.

To rerun a workflow, the source instance ID workflow actor will fork its
history state to the point of the target event ID, optionally update its
input, then send the new state to the target instance actor ID who will
then write that state and run the target activity, as well as  any
activity which are in-progress at that point in time. Care is taken to
ensure all currently active activities are executed if the target event
ID is in a fanout scenario.

```
rerunWorkfolow -> source Actor ID [Fork State] -{state}-> target Actor ID ->>> call activities
```

The workflow and activities actor code has been refactored to multiple
files for readability and maintainability.

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

Author: JoshVanL

go.mod

@@ -34,7 +34,7 @@ require (
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/hashicorp/raft v1.4.0
 	github.com/hashicorp/raft-boltdb v0.0.0-20230125174641-2a8082862702
-	github.com/jackc/pgx/v5 v5.5.5
+	github.com/jackc/pgx/v5 v5.7.4
 	github.com/jhump/protoreflect v1.15.3
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/lestrrat-go/jwx/v2 v2.0.21
@@ -307,8 +307,8 @@ require (
 	github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf // indirect
 	github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
-	github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9 // indirect
-	github.com/jackc/puddle/v2 v2.2.1 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
 	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
 	github.com/jcmturner/gofork v1.7.6 // indirect
@@ -517,3 +517,5 @@ 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-20250506134053-1cae3eb4b56b

go.sum

@@ -1,4 +1,4 @@
-  cel.dev/expr v0.19.2 h1:V354PbqIXr9IQdwy4SYA4xa0HXaWq1BUPAGzugBY5V4=
+cel.dev/expr v0.19.2 h1:V354PbqIXr9IQdwy4SYA4xa0HXaWq1BUPAGzugBY5V4=
 cel.dev/expr v0.19.2/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
@@ -504,8 +504,6 @@ github.com/danieljoos/wincred v1.1.2 h1:QLdCxFs1/Yl4zduvBdcHB8goaYk9RARS2SgLLRuA
 github.com/danieljoos/wincred v1.1.2/go.mod h1:GijpziifJoIBfYh+S7BbkdUTU4LfM+QnGqR5Vl2tAx0=
 github.com/dapr/components-contrib v1.15.1-0.20250423123324-b2c31ceba20b h1:u1xUF4CuNsTlv8pZsrX6XH1FolmMCjJ0Kyfef0wJkzY=
 github.com/dapr/components-contrib v1.15.1-0.20250423123324-b2c31ceba20b/go.mod h1:mDA9c8w/KGAOx18yCTaMXHkfWS6mFHR/o3FjI6a14dA=
-github.com/dapr/durabletask-go v0.6.5 h1:aWcxMfYudojpgRjJRdUr7yyZ7rGcvLtWXUuA4cGHBR0=
-github.com/dapr/durabletask-go v0.6.5/go.mod h1:nTZ5fCbJLnZbVdi6Z2YxdDF1OgQZL3LroogGuetrwuA=
 github.com/dapr/kit v0.15.2 h1:5H9IhKScU/SpE2Hxvr5vUlmYN1e2MJN15RoT8/KSziU=
 github.com/dapr/kit v0.15.2/go.mod h1:HwFsBKEbcyLanWlDZE7u/jnaDCD/tU+n3pkFNUctQNw=
 github.com/dave/jennifer v1.4.0/go.mod h1:fIb+770HOpJ2fmN9EPPKOqm1vMGhB+TwXKMZhrIygKg=
@@ -1069,12 +1067,12 @@ github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6
 github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
-github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9 h1:L0QtFUgDarD7Fpv9jeVMgy/+Ec0mtnmYuImjTz6dtDA=
-github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.5.5 h1:amBjrZVmksIdNjxGW/IiIMzxMKZFelXbUoPNb+8sjQw=
-github.com/jackc/pgx/v5 v5.5.5/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
-github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
-github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.7.4 h1:9wKznZrhWa2QiHL+NjTSPP6yjl3451BX3imWDnokYlg=
+github.com/jackc/pgx/v5 v5.7.4/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jawher/mow.cli v1.0.4/go.mod h1:5hQj2V8g+qYmLUVWqu4Wuja1pI57M83EChYLVZ0sMKk=
 github.com/jawher/mow.cli v1.2.0/go.mod h1:y+pcA3jBAdo/GIZx/0rFjw/K2bVEODP9rfZOfaiq8Ko=
 github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
@@ -1112,6 +1110,8 @@ 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/durabletask-go v0.0.0-20250506134053-1cae3eb4b56b h1:Shdwhif7Nr+1oJtrjEQJB+Tuaq2tLjiqBMtXaJGoKQM=
+github.com/joshvanl/durabletask-go v0.0.0-20250506134053-1cae3eb4b56b/go.mod h1:JhMyDybRUFmmgieGxCPeg9e2cWwtx4LwNXjD+LBtKYk=
 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/actors/router/router.go

@@ -110,18 +110,19 @@ func (r *router) Call(ctx context.Context, req *internalv1pb.InternalInvokeReque
 	var res *internalv1pb.InternalInvokeResponse
 	if r.resiliency.PolicyDefined(req.GetActor().GetActorType(), resiliency.ActorPolicy{}) {
 		res, err = r.callActor(ctx, req)
-		// Don't bubble perminant errors up to the caller to interfere with top level
-		// retries.
-		if _, ok := err.(*backoff.PermanentError); ok {
-			err = errors.Unwrap(err)
-		}
 	} else {
 		policyRunner := resiliency.NewRunner[*internalv1pb.InternalInvokeResponse](ctx, r.resiliency.BuiltInPolicy(resiliency.BuiltInActorNotFoundRetries))
 		res, err = policyRunner(func(ctx context.Context) (*internalv1pb.InternalInvokeResponse, error) {
 			return r.callActor(ctx, req)
 		})
 	}
 
+	// Don't bubble perminant errors up to the caller to interfere with top level
+	// retries.
+	if _, ok := err.(*backoff.PermanentError); ok {
+		err = errors.Unwrap(err)
+	}
+
 	return res, err
 }
 

pkg/actors/targets/activity/activity.go

@@ -0,0 +1,147 @@
+/*
+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 activity
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/dapr/dapr/pkg/actors"
+	actorapi "github.com/dapr/dapr/pkg/actors/api"
+	"github.com/dapr/dapr/pkg/actors/reminders"
+	"github.com/dapr/dapr/pkg/actors/router"
+	"github.com/dapr/dapr/pkg/actors/state"
+	"github.com/dapr/dapr/pkg/actors/table"
+	"github.com/dapr/dapr/pkg/actors/targets"
+	internalsv1pb "github.com/dapr/dapr/pkg/proto/internals/v1"
+	"github.com/dapr/dapr/pkg/runtime/wfengine/todo"
+	"github.com/dapr/kit/logger"
+)
+
+var log = logger.NewLogger("dapr.runtime.actors.targets.activity")
+
+type activity struct {
+	appID             string
+	actorID           string
+	actorType         string
+	workflowActorType string
+
+	table     table.Interface
+	router    router.Interface
+	state     state.Interface
+	reminders reminders.Interface
+
+	scheduler          todo.ActivityScheduler
+	reminderInterval   time.Duration
+	schedulerReminders bool
+}
+
+type ActivityOptions struct {
+	AppID              string
+	ActivityActorType  string
+	WorkflowActorType  string
+	ReminderInterval   *time.Duration
+	Scheduler          todo.ActivityScheduler
+	Actors             actors.Interface
+	SchedulerReminders bool
+}
+
+func ActivityFactory(ctx context.Context, opts ActivityOptions) (targets.Factory, error) {
+	table, err := opts.Actors.Table(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	router, err := opts.Actors.Router(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	state, err := opts.Actors.State(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	reminders, err := opts.Actors.Reminders(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return func(actorID string) targets.Interface {
+		reminderInterval := time.Minute * 1
+
+		if opts.ReminderInterval != nil {
+			reminderInterval = *opts.ReminderInterval
+		}
+
+		return &activity{
+			appID:              opts.AppID,
+			actorID:            actorID,
+			actorType:          opts.ActivityActorType,
+			workflowActorType:  opts.WorkflowActorType,
+			reminderInterval:   reminderInterval,
+			table:              table,
+			router:             router,
+			state:              state,
+			reminders:          reminders,
+			scheduler:          opts.Scheduler,
+			schedulerReminders: opts.SchedulerReminders,
+		}
+	}, nil
+}
+
+// InvokeMethod implements actors.InternalActor and schedules the background execution of a workflow activity.
+// Activities are scheduled by workflows and can execute for arbitrary lengths of time. Instead of executing
+// activity logic directly, InvokeMethod creates a reminder that executes the activity logic. InvokeMethod
+// returns immediately after creating the reminder, enabling the workflow to continue processing other events
+// in parallel.
+func (a *activity) InvokeMethod(ctx context.Context, req *internalsv1pb.InternalInvokeRequest) (*internalsv1pb.InternalInvokeResponse, error) {
+	return a.handleInvoke(ctx, req)
+}
+
+// InvokeReminder implements actors.InternalActor and executes the activity logic.
+func (a *activity) InvokeReminder(ctx context.Context, reminder *actorapi.Reminder) error {
+	return a.handleReminder(ctx, reminder)
+}
+
+// InvokeTimer implements actors.InternalActor
+func (a *activity) InvokeTimer(ctx context.Context, reminder *actorapi.Reminder) error {
+	return errors.New("timers are not implemented")
+}
+
+// DeactivateActor implements actors.InternalActor
+func (a *activity) Deactivate() error {
+	log.Debugf("Activity actor '%s': deactivated", a.actorID)
+	return nil
+}
+
+func (a *activity) InvokeStream(context.Context, *internalsv1pb.InternalInvokeRequest, chan<- *internalsv1pb.InternalInvokeResponse) error {
+	return errors.New("not implemented")
+}
+
+// Key returns the key for this unique actor.
+func (a *activity) Key() string {
+	return a.actorType + actorapi.DaprSeparator + a.actorID
+}
+
+// Type returns the type of actor.
+func (a *activity) Type() string {
+	return a.actorType
+}
+
+// ID returns the ID of the actor.
+func (a *activity) ID() string {
+	return a.actorID
+}

pkg/actors/targets/activity/execute.go

@@ -0,0 +1,132 @@
+/*
+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 activity
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+
+	diag "github.com/dapr/dapr/pkg/diagnostics"
+	invokev1 "github.com/dapr/dapr/pkg/messaging/v1"
+	internalsv1pb "github.com/dapr/dapr/pkg/proto/internals/v1"
+	wferrors "github.com/dapr/dapr/pkg/runtime/wfengine/errors"
+	"github.com/dapr/dapr/pkg/runtime/wfengine/todo"
+	"github.com/dapr/durabletask-go/api"
+	"github.com/dapr/durabletask-go/backend"
+)
+
+func (a *activity) executeActivity(ctx context.Context, name string, taskEvent *backend.HistoryEvent) (todo.RunCompleted, error) {
+	activityName := ""
+	if ts := taskEvent.GetTaskScheduled(); ts != nil {
+		activityName = ts.GetName()
+	} else {
+		return todo.RunCompletedTrue, fmt.Errorf("invalid activity task event: '%s'", taskEvent.String())
+	}
+
+	endIndex := strings.Index(a.actorID, "::")
+	if endIndex < 0 {
+		return todo.RunCompletedTrue, fmt.Errorf("invalid activity actor ID: '%s'", a.actorID)
+	}
+	workflowID := a.actorID[0:endIndex]
+
+	wi := &backend.ActivityWorkItem{
+		SequenceNumber: int64(taskEvent.GetEventId()),
+		InstanceID:     api.InstanceID(workflowID),
+		NewEvent:       taskEvent,
+		Properties:     make(map[string]any),
+	}
+
+	// Executing activity code is a one-way operation. We must wait for the app code to report its completion, which
+	// will trigger this callback channel.
+	// TODO: Need to come up with a design for timeouts. Some activities may need to run for hours but we also need
+	//       to handle the case where the app crashes and never responds to the workflow. It may be necessary to
+	//       introduce some kind of heartbeat protocol to help identify such cases.
+	callback := make(chan bool, 1)
+	wi.Properties[todo.CallbackChannelProperty] = callback
+	log.Debugf("Activity actor '%s': scheduling activity '%s' for workflow with instanceId '%s'", a.actorID, name, wi.InstanceID)
+	elapsed := float64(0)
+	start := time.Now()
+	err := a.scheduler(ctx, wi)
+	elapsed = diag.ElapsedSince(start)
+
+	if errors.Is(err, context.DeadlineExceeded) {
+		diag.DefaultWorkflowMonitoring.ActivityOperationEvent(ctx, activityName, diag.StatusRecoverable, elapsed)
+		return todo.RunCompletedFalse, wferrors.NewRecoverable(fmt.Errorf("timed-out trying to schedule an activity execution - this can happen if too many activities are running in parallel or if the workflow engine isn't running: %w", err))
+	} else if err != nil {
+		diag.DefaultWorkflowMonitoring.ActivityOperationEvent(ctx, activityName, diag.StatusRecoverable, elapsed)
+		return todo.RunCompletedFalse, wferrors.NewRecoverable(fmt.Errorf("failed to schedule an activity execution: %w", err))
+	}
+	diag.DefaultWorkflowMonitoring.ActivityOperationEvent(ctx, activityName, diag.StatusSuccess, elapsed)
+
+	// Activity execution started
+	start = time.Now()
+	executionStatus := ""
+	elapsed = float64(0)
+	// Record metrics on exit
+	defer func() {
+		if executionStatus != "" {
+			diag.DefaultWorkflowMonitoring.ActivityExecutionEvent(ctx, activityName, executionStatus, elapsed)
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		// Activity execution failed with recoverable error
+		elapsed = diag.ElapsedSince(start)
+		executionStatus = diag.StatusRecoverable
+		return todo.RunCompletedFalse, ctx.Err() // will be retried
+	case completed := <-callback:
+		elapsed = diag.ElapsedSince(start)
+		if !completed {
+			// Activity execution failed with recoverable error
+			executionStatus = diag.StatusRecoverable
+			return todo.RunCompletedFalse, wferrors.NewRecoverable(todo.ErrExecutionAborted) // AbandonActivityWorkItem was called
+		}
+	}
+	log.Debugf("Activity actor '%s': activity completed for workflow with instanceId '%s' activityName '%s'", a.actorID, wi.InstanceID, name)
+
+	// publish the result back to the workflow actor as a new event to be processed
+	resultData, err := proto.Marshal(wi.Result)
+	if err != nil {
+		// Returning non-recoverable error
+		executionStatus = diag.StatusFailed
+		return todo.RunCompletedTrue, err
+	}
+
+	req := internalsv1pb.
+		NewInternalInvokeRequest(todo.AddWorkflowEventMethod).
+		WithActor(a.workflowActorType, workflowID).
+		WithData(resultData).
+		WithContentType(invokev1.ProtobufContentType)
+
+	_, err = a.router.Call(ctx, req)
+	switch {
+	case err != nil:
+		// Returning recoverable error, record metrics
+		executionStatus = diag.StatusRecoverable
+		return todo.RunCompletedFalse, wferrors.NewRecoverable(fmt.Errorf("failed to invoke '%s' method on workflow actor: %w", todo.AddWorkflowEventMethod, err))
+	case wi.Result.GetTaskCompleted() != nil:
+		// Activity execution completed successfully
+		executionStatus = diag.StatusSuccess
+	case wi.Result.GetTaskFailed() != nil:
+		// Activity execution failed
+		executionStatus = diag.StatusFailed
+	}
+	return todo.RunCompletedTrue, nil
+}