mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zeekay
2bf2702045
Replace the coroutine-push activity path with deterministic, history-backed
replay. One way: the old push-result mechanism is removed.
Server (pkg/tasks):
- durable.go: seq-keyed activity records (wfact/ns/wf/run/seq) — the
exactly-once anchor. applyScheduleActivity is idempotent per (run,seq);
completeWorkflowActivity records the terminal outcome and schedules a new
workflow task so the worker replays and advances. Per-run keyedMutex
serializes advancing a run. Server-side retry+backoff per RetryPolicy
(temporal.RetryPolicy, finite default bound). Recover() rebuilds in-flight
dispatch from the store at boot (called from Embed).
- workflow tasks now carry the run's full event-sourced history.
- dispatch.go: ScheduleActivity/CompleteActivityTask(push) -> DispatchActivity
(seq-bound token) + ResolveActivityToken. workflowPeer/pendingActivity and
OpcodeDeliverActivityResult retired.
- respondWorkflowHandler applies ScheduleActivity{seq} commands under the run
lock; respondActivity* resolve the token and advance the run.
Worker (pkg/sdk/worker):
- workerEnv is now a replay decider: ExecuteActivity resolves from history by
seq (recorded result -> no re-dispatch; else emit ScheduleActivity{seq} and
end the episode via a blockedFuture). Stateless between episodes.
- removed OnActivityResult/pendingActivities push machinery.
Client (pkg/sdk/client): drop OnActivityResult + ScheduleActivity from
WorkerTransport (dead in the replay model).
Tests: e2e (exactly-once, real workers) green; new worker replay determinism
tests (unresolved->schedule+block, resolved->recorded result, 2-activity
determinism). Full 'go test -race ./pkg/tasks/... ./pkg/sdk/...' green.
113 lines
2.8 KiB
Go
113 lines
2.8 KiB
Go
// Copyright © 2026 Hanzo AI. MIT License.
|
|
|
|
package worker
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/hanzoai/tasks/pkg/sdk/client"
|
|
)
|
|
|
|
// startSubscriptions wires the worker's server-push handlers and
|
|
// subscribes once per kind. Called from Start. The transport pushes
|
|
// task deliveries asynchronously after Subscribe returns; each
|
|
// delivery dispatches in a fresh goroutine bounded by the worker's
|
|
// existing semaphores so concurrency caps still apply.
|
|
func (w *workerImpl) startSubscriptions() error {
|
|
ctx, cancel := w.ctxWithStop(context.Background())
|
|
defer cancel()
|
|
|
|
w.transport.OnWorkflowTask(func(task *client.WorkflowTask) {
|
|
w.dispatchPushedWorkflowTask(task)
|
|
})
|
|
w.transport.OnActivityTask(func(task *client.ActivityTask) {
|
|
w.dispatchPushedActivityTask(task)
|
|
})
|
|
|
|
wfReq := client.PollWorkflowTaskRequest{
|
|
Namespace: w.namespace,
|
|
TaskQueueName: w.taskQueue,
|
|
TaskQueueKind: 0,
|
|
Identity: w.identity,
|
|
}
|
|
wfSub, err := w.transport.SubscribeWorkflowTasks(ctx, wfReq)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
w.subMu.Lock()
|
|
w.workflowSubID = wfSub
|
|
w.subMu.Unlock()
|
|
|
|
actReq := client.PollActivityTaskRequest{
|
|
Namespace: w.namespace,
|
|
TaskQueueName: w.taskQueue,
|
|
TaskQueueKind: 0,
|
|
Identity: w.identity,
|
|
}
|
|
actSub, err := w.transport.SubscribeActivityTasks(ctx, actReq)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
w.subMu.Lock()
|
|
w.activitySubID = actSub
|
|
w.subMu.Unlock()
|
|
|
|
return nil
|
|
}
|
|
|
|
// dispatchPushedWorkflowTask runs in the transport's delivery goroutine
|
|
// and hands the task off to a worker-owned goroutine bounded by the
|
|
// workflowExecSem cap so back-pressure still applies. The transport
|
|
// goroutine returns immediately so it can ack the push and accept the
|
|
// next delivery.
|
|
func (w *workerImpl) dispatchPushedWorkflowTask(task *client.WorkflowTask) {
|
|
if w.stopped() {
|
|
return
|
|
}
|
|
w.wg.Add(1)
|
|
go func() {
|
|
defer w.wg.Done()
|
|
ctx, cancel := w.ctxWithStop(context.Background())
|
|
defer cancel()
|
|
|
|
if w.workflowExecSem != nil {
|
|
select {
|
|
case w.workflowExecSem <- struct{}{}:
|
|
case <-ctx.Done():
|
|
return
|
|
case <-w.stopCh:
|
|
return
|
|
}
|
|
defer func() { <-w.workflowExecSem }()
|
|
}
|
|
w.dispatchWorkflowTask(ctx, task)
|
|
}()
|
|
}
|
|
|
|
// dispatchPushedActivityTask is the activity counterpart. Rate limits
|
|
// are enforced before dispatch via activityLimiter and taskQueueLimiter,
|
|
// matching the contract the previous poll loop honoured.
|
|
func (w *workerImpl) dispatchPushedActivityTask(task *client.ActivityTask) {
|
|
if w.stopped() {
|
|
return
|
|
}
|
|
w.wg.Add(1)
|
|
go func() {
|
|
defer w.wg.Done()
|
|
ctx, cancel := w.ctxWithStop(context.Background())
|
|
defer cancel()
|
|
|
|
if w.activityLimiter != nil {
|
|
if err := w.activityLimiter.Wait(ctx); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if w.taskQueueLimiter != nil {
|
|
if err := w.taskQueueLimiter.Wait(ctx); err != nil {
|
|
return
|
|
}
|
|
}
|
|
w.dispatchActivityTask(ctx, task)
|
|
}()
|
|
}
|