refactor(dispatch): carry the principal by value, not by encoding

A dispatch token held its tenant as the canonical string, so delivering
its result had to parse one back and decide what to do when it did not
parse. Falling back to the root tenant there would have written one
tenant's result into another's shard.

The dispatcher runs in-process with the engine, so it holds the
Principal itself. There is nothing to encode, nothing to parse, and no
fallback to get wrong. The encoding stays where it is actually needed:
on a replication frame and a migration job, which leave the process.

Co-authored-by: Hanzo Dev <[email protected]>

pkg/tasks/dispatch.go

@@ -81,7 +81,7 @@ type subscription struct {
 type pendingWorkflowTask struct {
 	token        []byte
 	tokenStr     string
-	principal    string
+	principal    Principal
 	ns           string
 	queue        string
 	workflowID   string
@@ -95,7 +95,7 @@ type pendingWorkflowTask struct {
 type pendingActivityTask struct {
 	token            []byte
 	tokenStr         string
-	principal        string
+	principal        Principal
 	ns               string
 	queue            string
 	activityID       string
@@ -333,7 +333,7 @@ func (d *dispatcher) drainLocked(sub *subscription) {
 
 // EnqueueWorkflowTask creates a workflow task, mints its token, and
 // either delivers it to a subscribed peer or queues it for later.
-func (d *dispatcher) EnqueueWorkflowTask(principal, ns, queue, workflowID, runID, workflowType string, input []byte) {
+func (d *dispatcher) EnqueueWorkflowTask(principal Principal, ns, queue, workflowID, runID, workflowType string, input []byte) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
@@ -367,7 +367,7 @@ func (d *dispatcher) EnqueueWorkflowTask(principal, ns, queue, workflowID, runID
 // (ns, wf, run, seq) via ResolveActivityToken. activityID is deterministic
 // for (run, seq); the token is unique per dispatch so a re-dispatch (retry
 // / recovery) does not collide with a stale token.
-func (d *dispatcher) DispatchActivity(principal, ns, queue, workflowID, runID string, seq int, activityID, activityType string, input []byte, startToCloseMs, heartbeatMs int64) []byte {
+func (d *dispatcher) DispatchActivity(principal Principal, ns, queue, workflowID, runID string, seq int, activityID, activityType string, input []byte, startToCloseMs, heartbeatMs int64) []byte {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 

pkg/tasks/dispatch_test.go

@@ -65,7 +65,7 @@ func TestDispatcher_WorkflowAndActivityRoundTrip(t *testing.T) {
 
 	// Enqueue a workflow task. The dispatcher should immediately push
 	// OpcodeDeliverWorkflowTask to peerA.
-	d.EnqueueWorkflowTask("", ns, queue, workflowID, runID, wfType, []byte(`{"hello":"world"}`))
+	d.EnqueueWorkflowTask(Principal{}, ns, queue, workflowID, runID, wfType, []byte(`{"hello":"world"}`))
 
 	calls := cap.snapshot()
 	if len(calls) != 1 {
@@ -98,7 +98,7 @@ func TestDispatcher_WorkflowAndActivityRoundTrip(t *testing.T) {
 	// to the subscribed activity peer (peerB) and returns a token bound to
 	// (ns, wf, run, seq).
 	const actID = "run-1-act-0"
-	actToken := d.DispatchActivity("", ns, queue, workflowID, runID, 0, actID, actType,
+	actToken := d.DispatchActivity(Principal{}, ns, queue, workflowID, runID, 0, actID, actType,
 		[]byte(`{"name":"activity-input"}`), 60_000, 5_000)
 	if len(actToken) == 0 {
 		t.Fatalf("DispatchActivity returned empty token")
@@ -162,7 +162,7 @@ func TestDispatcher_QueuesUntilSubscribe(t *testing.T) {
 	d := newDispatcher()
 	d.send = cap.fn
 
-	d.EnqueueWorkflowTask("", "ns", "q", "wf", "run", "Type", nil)
+	d.EnqueueWorkflowTask(Principal{}, "ns", "q", "wf", "run", "Type", nil)
 	if got := len(cap.snapshot()); got != 0 {
 		t.Fatalf("unexpected push before subscribe: %d", got)
 	}
@@ -201,7 +201,7 @@ func TestDispatcher_PrunesDeadPeerOnSendFailure(t *testing.T) {
 		t.Fatalf("subscribe live: %v", err)
 	}
 
-	d.EnqueueWorkflowTask("", ns, queue, "wf", "run", "Type", nil)
+	d.EnqueueWorkflowTask(Principal{}, ns, queue, "wf", "run", "Type", nil)
 
 	mu.Lock()
 	got := append([]sendCall(nil), calls...)
@@ -230,7 +230,7 @@ func TestDispatcher_PrunesDeadPeerOnSendFailure(t *testing.T) {
 	mu.Lock()
 	calls = nil
 	mu.Unlock()
-	d.EnqueueWorkflowTask("", ns, queue, "wf2", "run2", "Type", nil)
+	d.EnqueueWorkflowTask(Principal{}, ns, queue, "wf2", "run2", "Type", nil)
 	mu.Lock()
 	got2 := append([]sendCall(nil), calls...)
 	mu.Unlock()

pkg/tasks/durable.go

@@ -178,7 +178,7 @@ func (e *engine) scheduleWorkflowTask(ns, wf, run string) error {
 		return err
 	}
 	if e.disp != nil {
-		e.disp.EnqueueWorkflowTask(e.principal.String(), ns, wfe.TaskQueue, wf, run, wfe.Type.Name, hist)
+		e.disp.EnqueueWorkflowTask(e.principal, ns, wfe.TaskQueue, wf, run, wfe.Type.Name, hist)
 	}
 	return nil
 }
@@ -250,7 +250,7 @@ func (e *engine) dispatchActivityAttempt(ns, wf, run string, rec *workflowActivi
 		}
 	}
 	if e.disp != nil {
-		e.disp.DispatchActivity(e.principal.String(), ns, rec.TaskQueue, wf, run, rec.Seq, rec.ActivityID, rec.ActivityType, rec.Input, rec.StartToCloseMs, rec.HeartbeatMs)
+		e.disp.DispatchActivity(e.principal, ns, rec.TaskQueue, wf, run, rec.Seq, rec.ActivityID, rec.ActivityType, rec.Input, rec.StartToCloseMs, rec.HeartbeatMs)
 	}
 	return nil
 }

pkg/tasks/embed.go

@@ -216,18 +216,6 @@ func (e *Embedded) Address() string {
 // route back to the owning shard via the task's org.
 type View struct{ en *engine }
 
-// mustPrincipal decodes a principal carried on a dispatch token. The token
-// was minted from a live engine view, so an undecodable one is a lost
-// tenant, not a caller error: fall back to the root tenant rather than drop
-// the delivery.
-func mustPrincipal(s string) Principal {
-	p, err := storepkg.ParsePrincipal(s)
-	if err != nil {
-		return Principal{}
-	}
-	return p
-}
-
 // Principal is the tenant a store view is scoped to: an org, optionally
 // narrowed to a project and a user. The zero value is the root tenant.
 type Principal = storepkg.Principal
@@ -2081,7 +2069,7 @@ func respondWorkflowHandler(en *engine) zap.Handler {
 		// Apply on the org that owns the run: org-scoped runs live in a prefixed
 		// store partition rootEn cannot see. The root principal (unscoped /
 		// loopback) makes As a no-op, so the in-process path is unchanged.
-		oe := en.As(mustPrincipal(t.principal))
+		oe := en.As(t.principal)
 		// Apply the decider's command batch. kind=2 (scheduleActivity)
 		// carries the deterministic seq + activity spec — this is where a
 		// workflow-driven activity enters the durable, event-sourced path.
@@ -2135,7 +2123,7 @@ func respondActivityCompletedHandler(en *engine) zap.Handler {
 		if !ok {
 			return objectAck(0, "task token not found", 404)
 		}
-		if err := en.As(mustPrincipal(pt.principal)).completeWorkflowActivity(pt.ns, pt.workflowID, pt.runID, pt.seq, result, nil); err != nil {
+		if err := en.As(pt.principal).completeWorkflowActivity(pt.ns, pt.workflowID, pt.runID, pt.seq, result, nil); err != nil {
 			return objectAck(0, err.Error(), 500)
 		}
 		return objectAck(0, "", 200)
@@ -2149,7 +2137,7 @@ func respondActivityFailedHandler(en *engine) zap.Handler {
 		if !ok {
 			return objectAck(0, "task token not found", 404)
 		}
-		if err := en.As(mustPrincipal(pt.principal)).completeWorkflowActivity(pt.ns, pt.workflowID, pt.runID, pt.seq, nil, failure); err != nil {
+		if err := en.As(pt.principal).completeWorkflowActivity(pt.ns, pt.workflowID, pt.runID, pt.seq, nil, failure); err != nil {
 			return objectAck(0, err.Error(), 500)
 		}
 		return objectAck(0, "", 200)