diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cc15021..0c99b9f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,7 +23,7 @@ jobs: ci: uses: xraph/workflows/.github/workflows/go-ci.yml@v1 with: - go-versions: '["1.26.9"]' + go-versions: '["1.26.10"]' os: '["ubuntu-latest"]' qualification: @@ -31,14 +31,14 @@ jobs: runs-on: ubuntu-latest env: GOWORK: "off" - GOTOOLCHAIN: go1.26.9 + GOTOOLCHAIN: go1.26.10 steps: - name: Checkout uses: actions/checkout@v6 - name: Set up Go uses: actions/setup-go@v6 with: - go-version: '1.26.9' + go-version: '1.26.10' cache-dependency-path: qualification/go.sum - name: Reject effective module replacements working-directory: qualification @@ -79,7 +79,7 @@ jobs: timeout-minutes: 15 env: GOWORK: "off" - GOTOOLCHAIN: go1.26.9 + GOTOOLCHAIN: go1.26.10 GOMAXPROCS: "2" WRITERFLOOR_REQUIRED: "1" WRITERFLOOR_TEST_DSN: postgres://dispatch:dispatch@127.0.0.1:5432/dispatch?sslmode=disable @@ -104,7 +104,7 @@ jobs: - uses: actions/checkout@v6 - uses: actions/setup-go@v6 with: - go-version: '1.26.9' + go-version: '1.26.10' cache-dependency-path: qualification/writerfloor/go.sum - name: Reject module replacements run: | @@ -207,7 +207,7 @@ jobs: timeout-minutes: 15 env: GOWORK: "off" - GOTOOLCHAIN: go1.26.9 + GOTOOLCHAIN: go1.26.10 GOMAXPROCS: "2" DISPATCH_LIFECYCLE_TEST_DSN: postgres://dispatch:dispatch@127.0.0.1:5432/dispatch?sslmode=disable services: @@ -229,7 +229,7 @@ jobs: - uses: actions/checkout@v6 - uses: actions/setup-go@v6 with: - go-version: '1.26.9' + go-version: '1.26.10' cache-dependency-path: | go.sum qualification/oldwriter/go.sum diff --git a/Makefile b/Makefile index 10bf95c..de521b5 100644 --- a/Makefile +++ b/Makefile @@ -255,14 +255,14 @@ i: install # overrides disabled so these checks exercise the pinned host composition. .PHONY: qualification-check qualification-lint qualification-lint: - cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.9 golangci-lint run --allow-serial-runners --max-same-issues=0 --max-issues-per-linter=0 ./... + cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.10 golangci-lint run --allow-serial-runners --max-same-issues=0 --max-issues-per-linter=0 ./... qualification-check: - cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.9 go mod tidy -diff - cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.9 go mod verify - cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.9 go vet ./... - cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.9 go test -race -count=1 ./... - cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.9 go build ./... + cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.10 go mod tidy -diff + cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.10 go mod verify + cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.10 go vet ./... + cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.10 go test -race -count=1 ./... + cd qualification && GOWORK=off GOTOOLCHAIN=go1.26.10 go build ./... .PHONY: qualification-process-check qualification-process-check: @@ -277,10 +277,10 @@ qualification-queryguard-check: @test -n "$$DISPATCH_LIFECYCLE_TEST_DSN" || { echo "DISPATCH_LIFECYCLE_TEST_DSN is required"; exit 1; } @set -eu; scratch=$$(mktemp -d "$${TMPDIR:-/tmp}/dispatch-queryguard.XXXXXX"); \ trap 'rm -rf "$$scratch"' EXIT; \ - (cd qualification/oldwriter && GOWORK=off GOTOOLCHAIN=go1.26.9 go build -p=1 -o "$$scratch/oldwriter" .); \ - (cd qualification/protocolone && GOWORK=off GOTOOLCHAIN=go1.26.9 go build -p=1 -o "$$scratch/protocolone" .); \ - (cd qualification/querypreguard && GOWORK=off GOTOOLCHAIN=go1.26.9 go build -p=1 -o "$$scratch/querypreguard" .); \ - GOWORK=off GOTOOLCHAIN=go1.26.9 DISPATCH_LIFECYCLE_REQUIRED=1 DISPATCH_QUERY_REQUIRED=1 \ + (cd qualification/oldwriter && GOWORK=off GOTOOLCHAIN=go1.26.10 go build -p=1 -o "$$scratch/oldwriter" .); \ + (cd qualification/protocolone && GOWORK=off GOTOOLCHAIN=go1.26.10 go build -p=1 -o "$$scratch/protocolone" .); \ + (cd qualification/querypreguard && GOWORK=off GOTOOLCHAIN=go1.26.10 go build -p=1 -o "$$scratch/querypreguard" .); \ + GOWORK=off GOTOOLCHAIN=go1.26.10 DISPATCH_LIFECYCLE_REQUIRED=1 DISPATCH_QUERY_REQUIRED=1 \ DISPATCH_OLDWRITER_BINARY="$$scratch/oldwriter" DISPATCH_PROTOCOL_ONE_BINARY="$$scratch/protocolone" \ DISPATCH_QUERYPREGUARD_BINARY="$$scratch/querypreguard" \ go test -p=1 -race -count=1 ./store/postgres -run 'TestQueryRemovalGuard|TestRetirementNativeOldWriter|TestQueryNativeProtocolOne' diff --git a/docs/architecture/durable-execution.md b/docs/architecture/durable-execution.md index 2fa9f8d..20f387c 100644 --- a/docs/architecture/durable-execution.md +++ b/docs/architecture/durable-execution.md @@ -135,7 +135,8 @@ activity handlers must observe it. Arbitrary blocking Go code cannot be preempte Workers renew grants while processing. Result publication and the workflow wakeup share one transaction. Unknown commit outcomes retry the same request, and an explicit revision conflict reloads the latest history. Polls filter by pinned build -inside the store. Engine health surfaces a failed durable worker. Polling is bounded +inside the store. Engine health surfaces a failed durable worker, and a worker +whose claims the store isn't answering. Polling is bounded and configurable; partition scheduling and long polling remain separate work. 2026-10-08: evaluator tests cover saved activity results and typed failures, diff --git a/docs/architecture/durable-worker-lifecycle.md b/docs/architecture/durable-worker-lifecycle.md index d93aae4..7974178 100644 --- a/docs/architecture/durable-worker-lifecycle.md +++ b/docs/architecture/durable-worker-lifecycle.md @@ -15,6 +15,28 @@ namespace-wide execution timeouts. `Claims` separates calls waiting on a claim response from calls processing their returned work. These counters belong to one Worker; they are not persisted build-retirement facts. +## Claims the store doesn't answer + +A claim can end without an answer: the store times out, the connection drops, or +the reply is lost after the grant committed. `Run` keeps polling through these. +Each poller backs off with full jitter, from `PollInterval` up to five seconds, +and starts over once a claim gets an answer. The compatibility read that gates +claims under retirement options works the same way. You'll see the error wrapped +in `ErrClaimUnknown` from `RunOnce`. + +Nothing was processed, so retrying can't run a task twice. A grant that committed +while its reply was lost keeps its lease, and the next claim after expiry takes a +new epoch, the same recovery a replacement worker relies on. The lost claim still +counts in `UnknownClaims`, so a drain can't report success. + +While a kind's claims keep failing, its `ClaimStatus` carries `Failures` and +`FailingSince`. `WorkerStatus.ClaimFailures` is the longest run among the kinds, +and `Engine.Health` fails for as long as one lasts. An error the store can't +classify, such as a missing table, is retried like a timeout, so watch health +for it. A claim the store refuses with a definitive error, such as `ErrInvalid` or +`ErrWriterCompatibility`, still stops `Run` unless it's ordinary lease contention. +So does any error while processing a claimed task. + ## Begin and observe a drain You supply a stable operation ID and explicit deadline: diff --git a/durable/runtime/worker.go b/durable/runtime/worker.go index 9a69b16..5ab69e3 100644 --- a/durable/runtime/worker.go +++ b/durable/runtime/worker.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "github.com/xraph/dispatch/backoff" "github.com/xraph/dispatch/durable" "github.com/xraph/dispatch/telemetry" ) @@ -150,6 +151,9 @@ func (w *Worker) StartExecution(ctx context.Context, request durable.StartReques // Run polls each task kind until cancellation or the first processing error. // Processing errors are returned to the supervisor, never silently discarded. +// A claim the store doesn't answer (ErrClaimUnknown) is retried with backoff +// and reported through Status, since nothing was processed and a task it may +// have taken returns to the queue when its lease expires. // Cancellation leaves unfinished tasks for replacement workers after expiry. func (w *Worker) Run(ctx context.Context) (result error) { // Run polls for as long as ctx lives. Whatever span ctx carries (a Forge @@ -170,11 +174,20 @@ func (w *Worker) Run(ctx context.Context) (result error) { for _, kind := range claimKinds { for range w.options.Concurrency { group.Go(func() { + retries := 0 for workCtx.Err() == nil { worked, err := w.runOnce(workCtx, kind, true) if errors.Is(err, ErrWorkerDraining) || errors.Is(err, ErrDrainIncomplete) { return } + if errors.Is(err, ErrClaimUnknown) { + retries = min(retries+1, maxClaimRetries) + if w.pollWait(workCtx, w.claimRetryDelay(retries)) != nil { + return + } + continue + } + retries = 0 if err != nil && !normalContention(err) { if workCtx.Err() == nil { select { @@ -186,7 +199,7 @@ func (w *Worker) Run(ctx context.Context) (result error) { } return } - if !worked && w.pollWait(workCtx) != nil { + if !worked && w.pollWait(workCtx, w.options.PollInterval) != nil { return } } @@ -216,11 +229,11 @@ func (w *Worker) runOnce(ctx context.Context, kind durable.TaskKind, runOwned bo defer func() { // Publish a supervisor-owned failure before dropping the final work // registration. A drain may be accepted before Run sees this return value. - failed := runOwned && err != nil && ctx.Err() == nil && !normalContention(err) + failed := runOwned && err != nil && ctx.Err() == nil && !normalContention(err) && !errors.Is(err, ErrClaimUnknown) w.leaveClaim(operation, err, failed) }() if compatibilityErr := w.checkRetirement(ctx); compatibilityErr != nil { - return false, compatibilityErr + return false, w.readinessReturned(ctx, operation, compatibilityErr) } if kind == TaskExecutionTimeout { @@ -237,7 +250,7 @@ func (w *Worker) runOnce(ctx context.Context, kind durable.TaskKind, runOwned bo Queue: w.options.Queue, BuildID: w.options.BuildID, Kind: kind, Owner: w.options.Owner, LeaseDuration: w.options.LeaseDuration}) }) - w.claimReturned(operation, err) + err = w.claimReturned(ctx, operation, err) w.recordPoll(kind, task != nil, err) if err != nil || task == nil { return false, err @@ -257,6 +270,20 @@ func (w *Worker) runOnce(ctx context.Context, kind durable.TaskKind, runOwned bo return true, err } +const ( + // maxClaimRetryDelay bounds how long a poller backs off while the store + // doesn't answer claims, so work resumes soon after the store recovers. + maxClaimRetryDelay = 5 * time.Second + maxClaimRetries = 32 +) + +// claimRetryDelay spreads retries with full jitter, so every poller of a +// worker, and every worker on the store, doesn't retry at once. +func (w *Worker) claimRetryDelay(attempt int) time.Duration { + limit := max(w.options.PollInterval, maxClaimRetryDelay) + return max(w.options.PollInterval, backoff.NewExponentialWithJitter(w.options.PollInterval, limit).Delay(attempt)) +} + func storeCall[T any](ctx context.Context, w *Worker, call func(context.Context) (T, error)) (T, error) { callCtx, cancel := context.WithTimeout(ctx, w.options.StoreTimeout) defer cancel() diff --git a/durable/runtime/worker_children.go b/durable/runtime/worker_children.go index 793b1a3..df5c2ee 100644 --- a/durable/runtime/worker_children.go +++ b/durable/runtime/worker_children.go @@ -117,7 +117,7 @@ func (w *Worker) runChildDelivery(ctx context.Context, operation *claimOperation delivery, err := storeCall(ctx, w, func(callCtx context.Context) (*durable.ChildDelivery, error) { return w.store.ClaimChildDelivery(callCtx, durable.ChildDeliveryClaimRequest{Namespace: w.options.Namespace, BuildID: w.options.BuildID, Owner: w.options.Owner, LeaseDuration: w.options.LeaseDuration}) }) - w.claimReturned(operation, err) + err = w.claimReturned(ctx, operation, err) w.recordPoll(TaskChildDelivery, delivery != nil, err) if err != nil || delivery == nil { return false, err diff --git a/durable/runtime/worker_claim_retry_test.go b/durable/runtime/worker_claim_retry_test.go new file mode 100644 index 0000000..b1c8811 --- /dev/null +++ b/durable/runtime/worker_claim_retry_test.go @@ -0,0 +1,127 @@ +package runtime_test + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/xraph/dispatch/durable" + drt "github.com/xraph/dispatch/durable/runtime" + "github.com/xraph/dispatch/store/memory" +) + +// claimTimeoutStore times out workflow claims the way Postgres does when +// its host is overloaded, until failures runs out. With commit set, each +// claim takes its task first, so only the reply is lost. +type claimTimeoutStore struct { + durable.Store + commit bool + failures atomic.Int64 +} + +func (s *claimTimeoutStore) ClaimTask(ctx context.Context, r durable.ClaimRequest) (*durable.Task, error) { + if r.Kind != durable.TaskWorkflow || s.failures.Add(-1) < 0 { + return s.Store.ClaimTask(ctx, r) + } + if s.commit { + if _, err := s.Store.ClaimTask(ctx, r); err != nil { + return nil, err + } + } + return nil, fmt.Errorf("dispatch/postgres: claim execution task: timeout: %w", context.DeadlineExceeded) +} + +func TestRunRetriesClaimsWithUnknownOutcome(t *testing.T) { + for _, tc := range []struct { + name string + commit bool + }{{"claim rolled back", false}, {"claim committed, reply lost", true}} { + t.Run(tc.name, func(t *testing.T) { + s := memory.New() + store := &claimTimeoutStore{Store: s, commit: tc.commit} + store.failures.Store(3) + options := workerOptions(t) + var calls atomic.Int64 + options.Workflows["order"] = func(w *drt.Workflow, _ []byte) ([]byte, error) { + return w.Activity("charge", "charge", "", nil).Get() + } + options.Activities["charge"] = func(context.Context, drt.ActivityInfo, []byte) ([]byte, error) { + calls.Add(1) + return []byte("paid"), nil + } + w := newWorker(t, store, options) + key := startWorkerRun(t, w, options) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + + for { + execution, err := s.GetExecution(ctx, key) + if err != nil { + t.Fatal(err) + } + if execution.State == durable.StateCompleted { + break + } + select { + case err = <-done: + t.Fatalf("Run stopped on a claim timeout: %v", err) + case <-ctx.Done(): + t.Fatalf("run never completed: %s", execution.State) + case <-time.After(time.Millisecond): + } + } + if calls.Load() != 1 { + t.Errorf("activity ran %d times", calls.Load()) + } + status := w.Status() + if status.State != drt.WorkerRunning || status.ClaimFailures != 0 || !status.ClaimFailingSince.IsZero() { + t.Errorf("status after recovery: %+v", status) + } + cancel() + if err := <-done; err != nil { + t.Fatalf("Run: %v", err) + } + }) + } +} + +// A claim error the store can't classify might never clear, so the worker +// keeps retrying but reports it, instead of failing or going quiet. +func TestRunReportsClaimsThatKeepFailing(t *testing.T) { + store := &claimTimeoutStore{Store: memory.New()} + store.failures.Store(1 << 40) + options := workerOptions(t) + options.Workflows["order"] = func(*drt.Workflow, []byte) ([]byte, error) { return nil, nil } + w := newWorker(t, store, options) + startWorkerRun(t, w, options) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + + for w.Status().ClaimFailures < 3 { + select { + case err := <-done: + t.Fatalf("Run stopped on a claim timeout: %v", err) + case <-ctx.Done(): + t.Fatal("claim failures were never reported") + case <-time.After(time.Millisecond): + } + } + status := w.Status() + if status.State != drt.WorkerRunning || status.ClaimFailingSince.IsZero() || status.UnknownClaims == 0 { + t.Errorf("status while claims fail: %+v", status) + } + if _, err := w.RunOnce(ctx, durable.TaskWorkflow); !errors.Is(err, drt.ErrClaimUnknown) || !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("RunOnce: %v", err) + } + cancel() + if err := <-done; err != nil { + t.Fatalf("Run: %v", err) + } +} diff --git a/durable/runtime/worker_execution_timeout.go b/durable/runtime/worker_execution_timeout.go index 8c8cc40..62eda0e 100644 --- a/durable/runtime/worker_execution_timeout.go +++ b/durable/runtime/worker_execution_timeout.go @@ -12,7 +12,7 @@ func (w *Worker) runExecutionTimeout(ctx context.Context, operation *claimOperat grant, err := storeCall(ctx, w, func(callCtx context.Context) (*durable.ExecutionTimeoutTask, error) { return w.store.ClaimExecutionTimeout(callCtx, durable.ExecutionTimeoutClaimRequest{Namespace: w.options.Namespace, Owner: w.options.Owner, LeaseDuration: w.options.LeaseDuration}) }) - w.claimReturned(operation, err) + err = w.claimReturned(ctx, operation, err) w.recordPoll(TaskExecutionTimeout, grant != nil, err) if err != nil || grant == nil { return false, err diff --git a/durable/runtime/worker_execution_timeout_test.go b/durable/runtime/worker_execution_timeout_test.go index 776222e..8f679e7 100644 --- a/durable/runtime/worker_execution_timeout_test.go +++ b/durable/runtime/worker_execution_timeout_test.go @@ -3,6 +3,7 @@ package runtime_test import ( "context" "errors" + "fmt" "reflect" "testing" "time" @@ -208,12 +209,24 @@ func TestWorkerExecutionTimeoutRejectsMalformedGrant(t *testing.T) { } } func TestWorkerExecutionTimeoutPollerFailure(t *testing.T) { - failure := errors.New("timeout claim unavailable") - s := &timeoutClaimStore{Store: memory.New(), failure: failure} + // A claim the store doesn't answer is retried and reported, not fatal. options := workerOptions(t) - ctx, cancel := context.WithTimeout(t.Context(), time.Second) + w := newWorker(t, &timeoutClaimStore{Store: memory.New(), failure: errors.New("timeout claim unavailable")}, options) + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) defer cancel() - if err := newWorker(t, s, options).Run(ctx); !errors.Is(err, failure) { + if err := w.Run(ctx); err != nil { + t.Fatalf("unanswered timeout claims stopped Run: %v", err) + } + for _, claims := range w.Status().Claims { + if claims.Kind == drt.TaskExecutionTimeout && (claims.Failures == 0 || claims.FailingSince.IsZero()) { + t.Fatalf("timeout poller hid failure: %+v", claims) + } + } + // A claim the store refuses still stops the worker. + rejected := fmt.Errorf("timeout claim rejected: %w", durable.ErrInvalid) + ctx, cancel = context.WithTimeout(t.Context(), time.Second) + defer cancel() + if err := newWorker(t, &timeoutClaimStore{Store: memory.New(), failure: rejected}, options).Run(ctx); !errors.Is(err, rejected) { t.Fatalf("timeout poller hid failure: %v", err) } } diff --git a/durable/runtime/worker_lifecycle.go b/durable/runtime/worker_lifecycle.go index 5dba9d3..a77e60c 100644 --- a/durable/runtime/worker_lifecycle.go +++ b/durable/runtime/worker_lifecycle.go @@ -3,6 +3,7 @@ package runtime import ( "context" "errors" + "fmt" "sync" "time" @@ -25,6 +26,10 @@ var ( ErrWorkerDraining = errors.New("durable runtime: worker claim admission closed") ErrDrainIncomplete = errors.New("durable runtime: drain incomplete") ErrDrainDeadline = errors.New("durable runtime: first drain acceptance deadline passed") + // ErrClaimUnknown wraps a claim the store didn't answer, such as a + // timeout. The task may be leased to this worker; it returns to the + // queue when the lease expires. Run retries these claims with backoff. + ErrClaimUnknown = errors.New("durable runtime: claim outcome unknown") ) // DrainRequest fixes the operation deadline independently of any observer. @@ -41,10 +46,14 @@ type DrainHandle struct { } // ClaimStatus counts local calls, including claims awaiting a database response. +// Failures counts this kind's consecutive ErrClaimUnknown claims since +// FailingSince; both reset when one of its claims gets an answer. type ClaimStatus struct { - Kind durable.TaskKind - Claiming int64 - Processing int64 + Kind durable.TaskKind + Claiming int64 + Processing int64 + Failures int64 + FailingSince time.Time } // WorkerStatus reports process facts. Zero local work does not prove a build @@ -58,8 +67,13 @@ type WorkerStatus struct { Claims []ClaimStatus UnknownClaims int64 Failure string - Drain *DrainHandle - ObservedAt time.Time + // ClaimFailures is the longest run of ErrClaimUnknown claims among the + // task kinds, and ClaimFailingSince when the oldest current run began. + // Claims has each kind's own run. + ClaimFailures int64 + ClaimFailingSince time.Time + Drain *DrainHandle + ObservedAt time.Time } // DrainResult separates the immutable operation outcome from current quiescence. @@ -74,6 +88,11 @@ type DrainResult struct { ObservedAt time.Time } +type claimFailure struct { + count int64 + since time.Time +} + type claimOperation struct { kind durable.TaskKind claiming bool @@ -100,12 +119,15 @@ type workerLifecycle struct { unknown int64 failure string drain *drainOperation + + // failing holds each kind whose last claim got no answer. + failing map[durable.TaskKind]claimFailure } var claimKinds = [...]durable.TaskKind{durable.TaskWorkflow, durable.TaskActivity, durable.TaskTimer, TaskTimeout, TaskChildDelivery, TaskExecutionTimeout} func newWorkerLifecycle() workerLifecycle { - return workerLifecycle{state: WorkerNotStarted, admission: make(chan struct{}), changed: make(chan struct{}), active: make(map[*claimOperation]struct{})} + return workerLifecycle{state: WorkerNotStarted, admission: make(chan struct{}), changed: make(chan struct{}), active: make(map[*claimOperation]struct{}), failing: make(map[durable.TaskKind]claimFailure)} } // Status returns a private snapshot and never calls storage or user code. @@ -118,7 +140,14 @@ func (w *Worker) Status() WorkerStatus { Ready: l.state == WorkerRunning && !l.closed, AdmissionClosed: l.closed, InFlight: int64(len(l.active)), UnknownClaims: l.unknown, Failure: l.failure, ObservedAt: time.Now().UTC()} for _, kind := range claimKinds { - entry := ClaimStatus{Kind: kind} + failure := l.failing[kind] + entry := ClaimStatus{Kind: kind, Failures: failure.count, FailingSince: failure.since} + if failure.count > 0 { + s.ClaimFailures = max(s.ClaimFailures, failure.count) + if s.ClaimFailingSince.IsZero() || failure.since.Before(s.ClaimFailingSince) { + s.ClaimFailingSince = failure.since + } + } for op := range l.active { if op.kind == kind { if op.claiming { @@ -342,7 +371,11 @@ func (w *Worker) enterClaim(ctx context.Context, kind durable.TaskKind) (context l.active[op] = struct{}{} return workCtx, op, nil } -func (w *Worker) claimReturned(op *claimOperation, err error) { + +// claimReturned ends a claim's store call. A claim the store didn't answer +// counts as unknown, so a drain can't report success, and comes back as +// ErrClaimUnknown so Run retries it. +func (w *Worker) claimReturned(ctx context.Context, op *claimOperation, err error) error { l := &w.lifecycle l.mu.Lock() defer l.mu.Unlock() @@ -350,6 +383,36 @@ func (w *Worker) claimReturned(op *claimOperation, err error) { if err != nil && !definitiveCommitError(err) { l.unknown++ } + return l.answered(ctx, op.kind, err) +} + +// readinessReturned classifies the compatibility read before a claim the +// same way, but nothing was claimed, so it adds no unknown claim. +func (w *Worker) readinessReturned(ctx context.Context, op *claimOperation, err error) error { + l := &w.lifecycle + l.mu.Lock() + defer l.mu.Unlock() + return l.answered(ctx, op.kind, err) +} + +// answered tracks a run of store calls that got no answer for kind, and +// wraps each in ErrClaimUnknown. A cancelled call is the worker stopping, +// not the store failing, so it isn't counted. The caller holds mu. +func (l *workerLifecycle) answered(ctx context.Context, kind durable.TaskKind, err error) error { + if err == nil || definitiveCommitError(err) { + delete(l.failing, kind) + return err + } + if ctx.Err() != nil { + return err + } + failure := l.failing[kind] + if failure.count == 0 { + failure.since = time.Now().UTC() + } + failure.count++ + l.failing[kind] = failure + return fmt.Errorf("%w: %w", ErrClaimUnknown, err) } func (w *Worker) leaveClaim(op *claimOperation, err error, failed bool) { op.cancel(nil) @@ -369,8 +432,8 @@ func (w *Worker) leaveClaim(op *claimOperation, err error, failed bool) { l.finishIdle() } -func (w *Worker) pollWait(ctx context.Context) error { - timer := time.NewTimer(w.options.PollInterval) +func (w *Worker) pollWait(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) defer timer.Stop() select { case <-ctx.Done(): diff --git a/durable/runtime/worker_readiness_test.go b/durable/runtime/worker_readiness_test.go index a92a1eb..534bba6 100644 --- a/durable/runtime/worker_readiness_test.go +++ b/durable/runtime/worker_readiness_test.go @@ -201,3 +201,35 @@ func TestRetirementReadinessBoundsStorageObservation(t *testing.T) { t.Fatalf("bounded readiness: %+v %v", r, err) } } + +// Unavailable compatibility is retried like an unanswered claim, while an +// incompatible deployment still stops the worker before any claim. +func TestRetirementRunRetriesUnavailableReadiness(t *testing.T) { + s, _, options := readinessFixture(t) + s.failure = context.DeadlineExceeded + options.PollInterval = time.Millisecond + w, err := drt.NewWorker(s, options) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + for s.inspections.Load() < 3 || w.Status().ClaimFailures == 0 { + select { + case err = <-done: + t.Fatalf("Run stopped on unavailable readiness: %v", err) + case <-ctx.Done(): + t.Fatal("readiness was not retried") + case <-time.After(time.Millisecond): + } + } + if status := w.Status(); status.State != drt.WorkerRunning || status.UnknownClaims != 0 || s.claims.Load() != 0 { + t.Fatalf("status while readiness is unavailable: %+v", status) + } + cancel() + if err = <-done; err != nil { + t.Fatalf("Run: %v", err) + } +} diff --git a/engine/durable.go b/engine/durable.go index 0687fae..c9666d0 100644 --- a/engine/durable.go +++ b/engine/durable.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "time" log "github.com/xraph/go-utils/log" @@ -200,6 +201,21 @@ func (eng *Engine) stopDurable(ctx context.Context) error { } } +// durableClaimError reports claims the store hasn't been answering. The +// worker keeps retrying them; this makes the outage visible meanwhile. +func (eng *Engine) durableClaimError() error { + worker := eng.DurableWorker() + if worker == nil { + return nil + } + status := worker.Status() + if status.ClaimFailures == 0 { + return nil + } + return fmt.Errorf("%w: %d consecutive claims failed since %s", drt.ErrClaimUnknown, + status.ClaimFailures, status.ClaimFailingSince.Format(time.RFC3339)) +} + func (eng *Engine) durableError() error { if eng.durable == nil { return nil diff --git a/engine/durable_claim_retry_test.go b/engine/durable_claim_retry_test.go new file mode 100644 index 0000000..0d27880 --- /dev/null +++ b/engine/durable_claim_retry_test.go @@ -0,0 +1,70 @@ +package engine_test + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/xraph/dispatch" + "github.com/xraph/dispatch/durable" + drt "github.com/xraph/dispatch/durable/runtime" + "github.com/xraph/dispatch/engine" + "github.com/xraph/dispatch/store/memory" +) + +// unreachableClaims times out every claim while down is set. +type unreachableClaims struct { + *memory.Store + down atomic.Bool +} + +func (s *unreachableClaims) ClaimTask(ctx context.Context, r durable.ClaimRequest) (*durable.Task, error) { + if s.down.Load() { + return nil, fmt.Errorf("dispatch/postgres: claim execution task: timeout: %w", context.DeadlineExceeded) + } + return s.Store.ClaimTask(ctx, r) +} + +func TestDurableEngineRidesOutClaimTimeouts(t *testing.T) { + s := &unreachableClaims{Store: memory.New()} + s.down.Store(true) + d, err := dispatch.New(dispatch.WithStore(s)) + if err != nil { + t.Fatal(err) + } + options := drt.Options{Namespace: t.Name(), Queue: "durable", BuildID: "v1", Owner: "worker", + PollInterval: time.Millisecond, Workflows: map[string]drt.WorkflowFunc{"order": func(*drt.Workflow, []byte) ([]byte, error) { return []byte("done"), nil }}} + eng, err := engine.Build(d, engine.WithDurableWorkflows(options)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = eng.Stop(ctx) + }) + request := durable.StartRequest{Key: durable.Key{Namespace: options.Namespace, WorkflowID: "order", RunID: "run"}, + RequestID: "start", WorkflowType: "order", BuildID: options.BuildID, Queue: options.Queue} + if _, err = eng.StartDurableWorkflow(t.Context(), request); err != nil { + t.Fatal(err) + } + if err = eng.Start(t.Context()); err != nil { + t.Fatal(err) + } + + waitFor(t, 2*time.Second, func() bool { + return errors.Is(eng.Health(t.Context()), drt.ErrClaimUnknown) + }) + s.down.Store(false) + waitFor(t, 5*time.Second, func() bool { + execution, getErr := s.GetExecution(t.Context(), request.Key) + return getErr == nil && execution.State == durable.StateCompleted + }) + waitFor(t, 5*time.Second, func() bool { return eng.Health(t.Context()) == nil }) + if status, _ := eng.DurableStatus(); status.State != drt.WorkerRunning { + t.Fatalf("worker state: %s", status.State) + } +} diff --git a/engine/engine.go b/engine/engine.go index be77cc2..e209f9e 100644 --- a/engine/engine.go +++ b/engine/engine.go @@ -744,10 +744,15 @@ func (eng *Engine) EnqueueRaw(ctx context.Context, name string, payload []byte, } // Health checks the health of the engine by pinging the dispatcher's store. +// It also fails while the durable worker has stopped or its claims go +// unanswered, since durable work isn't being picked up then. func (eng *Engine) Health(ctx context.Context) error { if err := eng.durableError(); err != nil { return fmt.Errorf("durable workflow worker: %w", err) } + if err := eng.durableClaimError(); err != nil { + return fmt.Errorf("durable workflow worker: %w", err) + } return eng.d.Store().Ping(ctx) } diff --git a/engine/lifecycle_test.go b/engine/lifecycle_test.go index 38ccefd..d1a32cf 100644 --- a/engine/lifecycle_test.go +++ b/engine/lifecycle_test.go @@ -3,6 +3,7 @@ package engine_test import ( "context" "errors" + "fmt" "reflect" "sync" "sync/atomic" @@ -374,7 +375,8 @@ func TestStopWorkersKeepsPublisherAndStoreLive(t *testing.T) { } func TestStopWorkersRetainsWorkerErrorAndFinalCleanup(t *testing.T) { - failure := errors.New("durable polling unavailable") + // Unanswered claims are retried; a refused one stops the worker. + failure := fmt.Errorf("durable polling rejected: %w", durable.ErrInvalid) s := &strictLifecycleStore{base: memory.New(), claimError: failure} d, err := dispatch.New(dispatch.WithStore(s), dispatch.WithConcurrency(1), dispatch.WithPollInterval(time.Millisecond)) if err != nil { diff --git a/go.mod b/go.mod index 079cd49..b10475f 100644 --- a/go.mod +++ b/go.mod @@ -2,7 +2,7 @@ module github.com/xraph/dispatch go 1.26.0 -toolchain go1.26.9 +toolchain go1.26.10 require ( github.com/jackc/pgx/v5 v5.11.0 diff --git a/qualification/README.md b/qualification/README.md index 04f952a..81ccda1 100644 --- a/qualification/README.md +++ b/qualification/README.md @@ -1,13 +1,13 @@ # Dispatch host qualification -Run the credential regression from the repository root with Go 1.26.9: +Run the credential regression from the repository root with Go 1.26.10: ```sh make qualification-lint qualification-check ``` This separate Go module consumes published Dispatch, Forge auth, Authsome and -Warden versions. You can run `GOWORK=off GOTOOLCHAIN=go1.26.9 go test -race +Warden versions. You can run `GOWORK=off GOTOOLCHAIN=go1.26.10 go test -race -count=1 ./...` from this directory. Keep local replacements out of `go.mod`; the CI job rejects them and checks the nested module independently. @@ -44,7 +44,7 @@ go install golang.org/x/vuln/cmd/govulncheck@latest make qualification-process-check ``` -You need Docker and Go 1.26.9. The gate builds the actual `cmd/sinkhost` executable, +You need Docker and Go 1.26.10. The gate builds the actual `cmd/sinkhost` executable, records its compiler/module metadata, scans the program and binary, and starts one private PostgreSQL 17 container capped at 512 MiB, one CPU and 128 PIDs. Chronicle, Relay, Dispatch and the webhook receiver run as separate native @@ -119,7 +119,7 @@ Build and start it from this directory: ```sh operator_dir=$(mktemp -d) -GOWORK=off GOTOOLCHAIN=go1.26.9 go build -o "$operator_dir/operatorhost" ./cmd/operatorhost +GOWORK=off GOTOOLCHAIN=go1.26.10 go build -o "$operator_dir/operatorhost" ./cmd/operatorhost "$operator_dir/operatorhost" --state-file "$operator_dir/state.json" & operator_pid=$! ``` @@ -293,7 +293,7 @@ with a private state directory and delayed worker activation: ```sh operator_dir=$(mktemp -d) -GOWORK=off GOTOOLCHAIN=go1.26.9 go build -o "$operator_dir/operatorhost" ./cmd/operatorhost +GOWORK=off GOTOOLCHAIN=go1.26.10 go build -o "$operator_dir/operatorhost" ./cmd/operatorhost "$operator_dir/operatorhost" --state-file "$operator_dir/state.json" \ --lifecycle-instance=deferral-demo --activation-file="$operator_dir/activate" & operator_pid=$! diff --git a/qualification/go.mod b/qualification/go.mod index d0ad14e..6c614c4 100644 --- a/qualification/go.mod +++ b/qualification/go.mod @@ -2,7 +2,7 @@ module github.com/xraph/dispatch/qualification go 1.26.0 -toolchain go1.26.9 +toolchain go1.26.10 require ( github.com/gobwas/ws v1.4.0 diff --git a/qualification/operator-check.sh b/qualification/operator-check.sh index 6c455bb..0505409 100644 --- a/qualification/operator-check.sh +++ b/qualification/operator-check.sh @@ -8,7 +8,7 @@ cleanup() { rm -rf "$operator_dir" } trap cleanup EXIT -export GOWORK=off GOTOOLCHAIN=go1.26.9 +export GOWORK=off GOTOOLCHAIN=go1.26.10 cd "$operator_root" # Metadata survives private binary and credential cleanup, including failed runs. diff --git a/qualification/process-check.sh b/qualification/process-check.sh index 24761f2..d3a2237 100755 --- a/qualification/process-check.sh +++ b/qualification/process-check.sh @@ -8,7 +8,7 @@ cleanup() { rm -rf "$process_dir" } trap cleanup EXIT -export GOWORK=off GOTOOLCHAIN=go1.26.9 +export GOWORK=off GOTOOLCHAIN=go1.26.10 cd "$root" # Metadata survives private binary and credential cleanup, including failed runs. diff --git a/qualification/writerfloor/Makefile b/qualification/writerfloor/Makefile index 48fd2eb..8bce1b2 100644 --- a/qualification/writerfloor/Makefile +++ b/qualification/writerfloor/Makefile @@ -1,5 +1,5 @@ export GOWORK := off -export GOTOOLCHAIN := go1.26.9 +export GOTOOLCHAIN := go1.26.10 export GOMAXPROCS := 2 export GOFLAGS := -p=1 diff --git a/qualification/writerfloor/QUALIFICATION.md b/qualification/writerfloor/QUALIFICATION.md index 7c999d8..0358609 100644 --- a/qualification/writerfloor/QUALIFICATION.md +++ b/qualification/writerfloor/QUALIFICATION.md @@ -10,7 +10,7 @@ drain. These runs used real Docker processes and PostgreSQL persistence. - Isolated Ubuntu 24.04.5 ARM64 VM, Linux 6.8, four CPUs and 6 GiB RAM. - Rootless Docker provider, separate rootful launcher and private mTLS channels. - PostgreSQL 17.11, capped at 384 MiB and one CPU, with durable writes enabled. -- Go 1.26.9 with workspace overrides disabled and no module replacements. +- Go 1.26.10 with workspace overrides disabled and no module replacements. - Dispatch `de3621a604dd`, Ctrlplane `87291467f279`, and the published qualification module at `e43a3e4229e0`. Exact versions are retained in `go.mod`. - Test executable SHA256: diff --git a/qualification/writerfloor/README.md b/qualification/writerfloor/README.md index f4db8ff..e027c96 100644 --- a/qualification/writerfloor/README.md +++ b/qualification/writerfloor/README.md @@ -22,7 +22,7 @@ The current snapshot validates the complete finite delegation chain before mutat ## Checks -Resolve the published module graph with `GOWORK=off GOTOOLCHAIN=go1.26.9 go mod tidy`. There are no replacements. The Ctrlplane pin includes the published protected marker and Docker custody implementation. The pin also includes guarded recovery and independently retained provision receipts. The assembled Linux campaign below exercises the complete guarded recovery path. +Resolve the published module graph with `GOWORK=off GOTOOLCHAIN=go1.26.10 go mod tidy`. There are no replacements. The Ctrlplane pin includes the published protected marker and Docker custody implementation. The pin also includes guarded recovery and independently retained provision receipts. The assembled Linux campaign below exercises the complete guarded recovery path. Use a disposable PostgreSQL server, then run `make f`, `make l` and `WRITERFLOOR_TEST_DSN=... make check`. Tests create a fresh schema and clean it up. `WRITERFLOOR_REQUIRED=1` turns a missing server into a failure; the Makefile sets it for the test target. @@ -73,7 +73,7 @@ separate rootful launcher. Never point these tests at a shared or production server. They create disposable databases, restart processes, remove their own containers and networks, restore backups and promote a physical replica. -Build the service image from `cmd/managed-lifecycle-qualify` with Go 1.26.9, +Build the service image from `cmd/managed-lifecycle-qualify` with Go 1.26.10, `GOWORK=off`, `GOOS=linux`, the VM's `GOARCH` and `CGO_ENABLED=0`. Install it as `/app/workload` in a scratch image running as UID 10001, then push to your isolated registry and select its exact repository digest. Workers use the configured diff --git a/qualification/writerfloor/scripts/build_live_tests.py b/qualification/writerfloor/scripts/build_live_tests.py index 6ca901c..ebcfdc5 100644 --- a/qualification/writerfloor/scripts/build_live_tests.py +++ b/qualification/writerfloor/scripts/build_live_tests.py @@ -33,7 +33,7 @@ def main(): output = args.output.absolute() output.mkdir(mode=0o700) module = Path(__file__).resolve().parents[1] - env = dict(os.environ, GOWORK="off", GOTOOLCHAIN="go1.26.9", GOOS="linux", + env = dict(os.environ, GOWORK="off", GOTOOLCHAIN="go1.26.10", GOOS="linux", GOARCH=args.arch, CGO_ENABLED="1" if args.race else "0", GOMAXPROCS="2", GOFLAGS="-p=1") binary = output / "managedhost.test"