002 phase 1.1+1.2: Claim carries WorkerID; fencing calls take *Claim

Makefile

@@ -1,13 +1,28 @@
 
-.PHONY: test
+EXPERIMENTAL_MODULES := \
+	experimental/worker \
+	experimental/store/postgres \
+	experimental/store/sqlite
+
+.PHONY: all test cover test-experimental test-all clean
+
+all: test-all
+
 test:
 	go test . ./activities ./script ./workflowtest
 
-.PHONY: cover
 cover:
 	go test -coverprofile cover.out . ./activities ./script ./workflowtest
 	go tool cover -html=cover.out
 
-.PHONY: test-all
-test-all: test
+test-experimental:
+	@set -e; for mod in $(EXPERIMENTAL_MODULES); do \
+		echo "==> $$mod"; \
+		(cd $$mod && go build ./... && go vet ./... && go test ./...); \
+	done
+
+test-all: test test-experimental
 	go vet ./...
+
+clean:
+	rm -f cover.out

experimental/store/postgres/checkpointer.go

@@ -18,25 +18,25 @@ import (
 //
 // Reads (LoadCheckpoint) are unfenced: a fresh attempt must be able
 // to resume regardless of which worker originally wrote the snapshot.
-func (s *Store) NewCheckpointer(lease worker.Lease) workflow.Checkpointer {
-	return &leasedCheckpointer{store: s, lease: lease}
+func (s *Store) NewCheckpointer(claim *worker.Claim) workflow.Checkpointer {
+	return &leasedCheckpointer{store: s, claim: claim}
 }
 
 type leasedCheckpointer struct {
 	store *Store
-	lease worker.Lease
+	claim *worker.Claim
 }
 
 // SaveCheckpoint marshals the checkpoint to JSON and writes it to the
 // workflow_runs.checkpoint column under (claimed_by, attempt) fencing.
-// The checkpoint's ExecutionID must match the lease's RunID.
+// The checkpoint's ExecutionID must match the claim's run ID.
 func (c *leasedCheckpointer) SaveCheckpoint(ctx context.Context, checkpoint *workflow.Checkpoint) error {
 	if checkpoint == nil {
 		return fmt.Errorf("postgres: nil checkpoint")
 	}
-	if checkpoint.ExecutionID != c.lease.RunID {
-		return fmt.Errorf("postgres: checkpoint execution ID %q does not match lease run ID %q",
-			checkpoint.ExecutionID, c.lease.RunID)
+	if checkpoint.ExecutionID != c.claim.ID {
+		return fmt.Errorf("postgres: checkpoint execution ID %q does not match claim run ID %q",
+			checkpoint.ExecutionID, c.claim.ID)
 	}
 	blob, err := json.Marshal(checkpoint)
 	if err != nil {
@@ -48,9 +48,9 @@ func (c *leasedCheckpointer) SaveCheckpoint(ctx context.Context, checkpoint *wor
 		WHERE id         = $2
 		  AND claimed_by = $3
 		  AND attempt    = $4
-	`, blob, c.lease.RunID, c.lease.WorkerID, c.lease.Attempt)
+	`, blob, c.claim.ID, c.claim.WorkerID, c.claim.Attempt)
 	if err != nil {
-		return fmt.Errorf("postgres: save checkpoint %s: %w", c.lease.RunID, err)
+		return fmt.Errorf("postgres: save checkpoint %s: %w", c.claim.ID, err)
 	}
 	if tag.RowsAffected() == 0 {
 		return worker.ErrLeaseLost
@@ -95,9 +95,9 @@ func (c *leasedCheckpointer) DeleteCheckpoint(ctx context.Context, executionID s
 		WHERE id         = $1
 		  AND claimed_by = $2
 		  AND attempt    = $3
-	`, c.lease.RunID, c.lease.WorkerID, c.lease.Attempt)
+	`, c.claim.ID, c.claim.WorkerID, c.claim.Attempt)
 	if err != nil {
-		return fmt.Errorf("postgres: delete checkpoint %s: %w", c.lease.RunID, err)
+		return fmt.Errorf("postgres: delete checkpoint %s: %w", c.claim.ID, err)
 	}
 	if tag.RowsAffected() == 0 {
 		return worker.ErrLeaseLost

experimental/store/postgres/queue.go

@@ -85,6 +85,7 @@ func (s *Store) ClaimQueued(ctx context.Context, workerID string) (*worker.Claim
 		ID:           id,
 		Spec:         spec,
 		Attempt:      newAttempt,
+		WorkerID:     workerID,
 		OrgID:        orgID,
 		WorkflowType: workflowType,
 		CreditCost:   creditCost,
@@ -95,17 +96,17 @@ func (s *Store) ClaimQueued(ctx context.Context, workerID string) (*worker.Claim
 // Heartbeat implements worker.QueueStore with (claimed_by, attempt)
 // fencing. Rows with a status other than running, or a mismatched
 // lease, produce ErrLeaseLost.
-func (s *Store) Heartbeat(ctx context.Context, lease worker.Lease) error {
+func (s *Store) Heartbeat(ctx context.Context, claim *worker.Claim) error {
 	tag, err := s.pool.Exec(ctx, `
 		UPDATE workflow_runs
 		SET heartbeat_at = NOW()
 		WHERE id         = $1
 		  AND claimed_by = $2
 		  AND attempt    = $3
 		  AND status     = $4
-	`, lease.RunID, lease.WorkerID, lease.Attempt, string(worker.StatusRunning))
+	`, claim.ID, claim.WorkerID, claim.Attempt, string(worker.StatusRunning))
 	if err != nil {
-		return fmt.Errorf("postgres: heartbeat %s: %w", lease.RunID, err)
+		return fmt.Errorf("postgres: heartbeat %s: %w", claim.ID, err)
 	}
 	if tag.RowsAffected() == 0 {
 		return worker.ErrLeaseLost
@@ -114,7 +115,7 @@ func (s *Store) Heartbeat(ctx context.Context, lease worker.Lease) error {
 }
 
 // Complete implements worker.QueueStore with (claimed_by, attempt) fencing.
-func (s *Store) Complete(ctx context.Context, lease worker.Lease, outcome worker.Outcome) error {
+func (s *Store) Complete(ctx context.Context, claim *worker.Claim, outcome worker.Outcome) error {
 	tag, err := s.pool.Exec(ctx, `
 		UPDATE workflow_runs
 		SET status        = $1,
@@ -128,14 +129,14 @@ func (s *Store) Complete(ctx context.Context, lease worker.Lease, outcome worker
 		string(outcome.Status),
 		outcome.Result,
 		outcome.ErrorMessage,
-		lease.RunID,
-		lease.WorkerID,
+		claim.ID,
+		claim.WorkerID,
 		string(worker.StatusCompleted),
 		string(worker.StatusFailed),
-		lease.Attempt,
+		claim.Attempt,
 	)
 	if err != nil {
-		return fmt.Errorf("postgres: complete %s: %w", lease.RunID, err)
+		return fmt.Errorf("postgres: complete %s: %w", claim.ID, err)
 	}
 	if tag.RowsAffected() == 0 {
 		return worker.ErrLeaseLost

experimental/store/postgres/store_test.go

@@ -70,6 +70,9 @@ func TestStore_EnqueueClaimCompleteRoundTrip(t *testing.T) {
 	if claim.ID != "run-a" || claim.Attempt != 1 || string(claim.Spec) != `{"type":"demo"}` {
 		t.Fatalf("unexpected claim: %+v", claim)
 	}
+	if claim.WorkerID != "test-worker" {
+		t.Fatalf("claim.WorkerID = %q, want %q", claim.WorkerID, "test-worker")
+	}
 
 	again, err := store.ClaimQueued(ctx, "test-worker")
 	if err != nil {
@@ -79,12 +82,11 @@ func TestStore_EnqueueClaimCompleteRoundTrip(t *testing.T) {
 		t.Fatalf("expected no more queued runs, got %+v", again)
 	}
 
-	lease := worker.Lease{RunID: claim.ID, WorkerID: "test-worker", Attempt: claim.Attempt}
-	if err := store.Heartbeat(ctx, lease); err != nil {
+	if err := store.Heartbeat(ctx, claim); err != nil {
 		t.Fatalf("heartbeat: %v", err)
 	}
 
-	if err := store.Complete(ctx, lease, worker.Outcome{
+	if err := store.Complete(ctx, claim, worker.Outcome{
 		Status: worker.StatusCompleted,
 		Result: []byte(`{"ok":true}`),
 	}); err != nil {
@@ -96,17 +98,24 @@ func TestStore_HeartbeatLeaseLost(t *testing.T) {
 	store, _ := openTestStore(t)
 	ctx := context.Background()
 
-	_ = store.Enqueue(ctx, worker.NewRun{ID: "run-b"})
-	claim, _ := store.ClaimQueued(ctx, "alpha")
+	if err := store.Enqueue(ctx, worker.NewRun{ID: "run-b", Spec: []byte(`{}`)}); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	claim, err := store.ClaimQueued(ctx, "alpha")
+	if err != nil || claim == nil {
+		t.Fatalf("claim: %v / %+v", err, claim)
+	}
 
 	// Wrong worker ID should be rejected.
-	wrong := worker.Lease{RunID: claim.ID, WorkerID: "beta", Attempt: claim.Attempt}
-	if err := store.Heartbeat(ctx, wrong); err != worker.ErrLeaseLost {
+	wrong := *claim
+	wrong.WorkerID = "beta"
+	if err := store.Heartbeat(ctx, &wrong); err != worker.ErrLeaseLost {
 		t.Fatalf("expected ErrLeaseLost, got %v", err)
 	}
 	// Wrong attempt should be rejected.
-	badAttempt := worker.Lease{RunID: claim.ID, WorkerID: "alpha", Attempt: 99}
-	if err := store.Heartbeat(ctx, badAttempt); err != worker.ErrLeaseLost {
+	badAttempt := *claim
+	badAttempt.Attempt = 99
+	if err := store.Heartbeat(ctx, &badAttempt); err != worker.ErrLeaseLost {
 		t.Fatalf("expected ErrLeaseLost, got %v", err)
 	}
 }
@@ -115,8 +124,8 @@ func TestStore_ReclaimAndDeadLetter(t *testing.T) {
 	store, pool := openTestStore(t)
 	ctx := context.Background()
 
-	_ = store.Enqueue(ctx, worker.NewRun{ID: "reclaim-me"})
-	_ = store.Enqueue(ctx, worker.NewRun{ID: "dead-letter-me"})
+	_ = store.Enqueue(ctx, worker.NewRun{ID: "reclaim-me", Spec: []byte(`{}`)})
+	_ = store.Enqueue(ctx, worker.NewRun{ID: "dead-letter-me", Spec: []byte(`{}`)})
 
 	c1, _ := store.ClaimQueued(ctx, "w")
 	c2, _ := store.ClaimQueued(ctx, "w")
@@ -157,14 +166,10 @@ func TestStore_CheckpointerRoundTrip(t *testing.T) {
 	store, _ := openTestStore(t)
 	ctx := context.Background()
 
-	_ = store.Enqueue(ctx, worker.NewRun{ID: "cp-run"})
+	_ = store.Enqueue(ctx, worker.NewRun{ID: "cp-run", Spec: []byte(`{}`)})
 	claim, _ := store.ClaimQueued(ctx, "w")
 
-	cp := store.NewCheckpointer(worker.Lease{
-		RunID:    claim.ID,
-		WorkerID: "w",
-		Attempt:  claim.Attempt,
-	})
+	cp := store.NewCheckpointer(claim)
 
 	// Load before save -> ErrNoCheckpoint.
 	if _, err := cp.LoadCheckpoint(ctx, claim.ID); err != workflow.ErrNoCheckpoint {
@@ -192,9 +197,9 @@ func TestStore_CheckpointerRoundTrip(t *testing.T) {
 	}
 
 	// Wrong lease -> ErrLeaseLost.
-	bogus := store.NewCheckpointer(worker.Lease{
-		RunID: claim.ID, WorkerID: "someone-else", Attempt: claim.Attempt,
-	})
+	otherWorker := *claim
+	otherWorker.WorkerID = "someone-else"
+	bogus := store.NewCheckpointer(&otherWorker)
 	if err := bogus.SaveCheckpoint(ctx, original); err != worker.ErrLeaseLost {
 		t.Fatalf("expected ErrLeaseLost, got %v", err)
 	}
@@ -212,7 +217,7 @@ func TestStore_StepProgressAndActivityLog(t *testing.T) {
 	store, _ := openTestStore(t)
 	ctx := context.Background()
 
-	_ = store.Enqueue(ctx, worker.NewRun{ID: "obs-run"})
+	_ = store.Enqueue(ctx, worker.NewRun{ID: "obs-run", Spec: []byte(`{}`)})
 
 	// Step progress upsert.
 	p := workflow.StepProgress{

experimental/store/sqlite/checkpointer.go

@@ -12,18 +12,18 @@ import (
 
 type leasedCheckpointer struct {
 	store *Store
-	lease worker.Lease
+	claim *worker.Claim
 }
 
 // SaveCheckpoint implements workflow.Checkpointer. The checkpoint's
-// ExecutionID must match the lease's RunID.
+// ExecutionID must match the claim's run ID.
 func (c *leasedCheckpointer) SaveCheckpoint(ctx context.Context, checkpoint *workflow.Checkpoint) error {
 	if checkpoint == nil {
 		return fmt.Errorf("sqlite: nil checkpoint")
 	}
-	if checkpoint.ExecutionID != c.lease.RunID {
-		return fmt.Errorf("sqlite: checkpoint execution ID %q does not match lease run ID %q",
-			checkpoint.ExecutionID, c.lease.RunID)
+	if checkpoint.ExecutionID != c.claim.ID {
+		return fmt.Errorf("sqlite: checkpoint execution ID %q does not match claim run ID %q",
+			checkpoint.ExecutionID, c.claim.ID)
 	}
 	blob, err := json.Marshal(checkpoint)
 	if err != nil {
@@ -35,9 +35,9 @@ func (c *leasedCheckpointer) SaveCheckpoint(ctx context.Context, checkpoint *wor
 		WHERE id         = ?
 		  AND claimed_by = ?
 		  AND attempt    = ?
-	`, blob, c.lease.RunID, c.lease.WorkerID, c.lease.Attempt)
+	`, blob, c.claim.ID, c.claim.WorkerID, c.claim.Attempt)
 	if err != nil {
-		return fmt.Errorf("sqlite: save checkpoint %s: %w", c.lease.RunID, err)
+		return fmt.Errorf("sqlite: save checkpoint %s: %w", c.claim.ID, err)
 	}
 	n, _ := result.RowsAffected()
 	if n == 0 {
@@ -73,15 +73,19 @@ func (c *leasedCheckpointer) LoadCheckpoint(ctx context.Context, executionID str
 }
 
 // DeleteCheckpoint implements workflow.Checkpointer with (claimed_by,
-// attempt) fencing.
+// attempt) fencing. The executionID must match the claim's run ID.
 func (c *leasedCheckpointer) DeleteCheckpoint(ctx context.Context, executionID string) error {
+	if executionID != c.claim.ID {
+		return fmt.Errorf("sqlite: execution ID %q does not match claim run ID %q",
+			executionID, c.claim.ID)
+	}
 	result, err := c.store.db.ExecContext(ctx, `
 		UPDATE workflow_runs
 		SET checkpoint = NULL
 		WHERE id         = ?
 		  AND claimed_by = ?
 		  AND attempt    = ?
-	`, executionID, c.lease.WorkerID, c.lease.Attempt)
+	`, executionID, c.claim.WorkerID, c.claim.Attempt)
 	if err != nil {
 		return fmt.Errorf("sqlite: delete checkpoint %s: %w", executionID, err)
 	}

experimental/store/sqlite/queue.go

@@ -70,6 +70,7 @@ func (s *Store) ClaimQueued(ctx context.Context, workerID string) (*worker.Claim
 		ID:           id,
 		Spec:         spec,
 		Attempt:      attempt,
+		WorkerID:     workerID,
 		OrgID:        orgID,
 		WorkflowType: workflowType,
 		CreditCost:   creditCost,
@@ -78,7 +79,7 @@ func (s *Store) ClaimQueued(ctx context.Context, workerID string) (*worker.Claim
 }
 
 // Heartbeat implements worker.QueueStore.
-func (s *Store) Heartbeat(ctx context.Context, lease worker.Lease) error {
+func (s *Store) Heartbeat(ctx context.Context, claim *worker.Claim) error {
 	now := time.Now().UTC().Format(timeFormat)
 	result, err := s.db.ExecContext(ctx, `
 		UPDATE workflow_runs
@@ -87,9 +88,9 @@ func (s *Store) Heartbeat(ctx context.Context, lease worker.Lease) error {
 		  AND claimed_by = ?
 		  AND attempt    = ?
 		  AND status     = ?
-	`, now, lease.RunID, lease.WorkerID, lease.Attempt, string(worker.StatusRunning))
+	`, now, claim.ID, claim.WorkerID, claim.Attempt, string(worker.StatusRunning))
 	if err != nil {
-		return fmt.Errorf("sqlite: heartbeat %s: %w", lease.RunID, err)
+		return fmt.Errorf("sqlite: heartbeat %s: %w", claim.ID, err)
 	}
 	n, _ := result.RowsAffected()
 	if n == 0 {
@@ -99,7 +100,7 @@ func (s *Store) Heartbeat(ctx context.Context, lease worker.Lease) error {
 }
 
 // Complete implements worker.QueueStore.
-func (s *Store) Complete(ctx context.Context, lease worker.Lease, outcome worker.Outcome) error {
+func (s *Store) Complete(ctx context.Context, claim *worker.Claim, outcome worker.Outcome) error {
 	completedAt := nullableTime(time.Time{})
 	if outcome.Status == worker.StatusCompleted || outcome.Status == worker.StatusFailed {
 		completedAt = nullableTime(time.Now())
@@ -118,12 +119,12 @@ func (s *Store) Complete(ctx context.Context, lease worker.Lease, outcome worker
 		outcome.Result,
 		outcome.ErrorMessage,
 		completedAt,
-		lease.RunID,
-		lease.WorkerID,
-		lease.Attempt,
+		claim.ID,
+		claim.WorkerID,
+		claim.Attempt,
 	)
 	if err != nil {
-		return fmt.Errorf("sqlite: complete %s: %w", lease.RunID, err)
+		return fmt.Errorf("sqlite: complete %s: %w", claim.ID, err)
 	}
 	n, _ := result.RowsAffected()
 	if n == 0 {

experimental/store/sqlite/store.go

@@ -58,9 +58,12 @@ func (s *Store) Migrate(ctx context.Context) error {
 }
 
 // NewCheckpointer returns a lease-fenced Checkpointer for the given
-// claim.
-func (s *Store) NewCheckpointer(lease worker.Lease) *leasedCheckpointer {
-	return &leasedCheckpointer{store: s, lease: lease}
+// claim. Panics if claim is nil.
+func (s *Store) NewCheckpointer(claim *worker.Claim) *leasedCheckpointer {
+	if claim == nil {
+		panic("sqlite: nil claim")
+	}
+	return &leasedCheckpointer{store: s, claim: claim}
 }
 
 func formatTime(t time.Time) string {