mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zeekay
38e44b507d
pkg/sdk/examples/{hello,signal,cron}.go drive the engine end-to-end
through pkg/sdk/client over the ZAP wire — zero HTTP, zero
go.temporal.io. Tests boot tasks.Embed in-process, dial the local
ZAP node, and round-trip every workflow + schedule opcode.
Engine fixes (pkg/tasks/embed.go):
- ZAP handlers now read the SDK's snake_case JSON shape
(workflow_id / workflow_type / task_queue / signal_name / schedule_id)
in addition to the camelCase shape used by the HTTP shim
- DescribeWorkflow returns {info: WorkflowExecutionInfo} (SDK shape)
- WorkflowExecutionInfo emits status as int8 (1=RUNNING / 4=CANCELED / 5=TERMINATED)
- StartWorkflow returns {run_id, workflow_id}
- PauseSchedule respects the requested paused flag (was always true)
- Schedule list response goes through scheduleSDK marshaller
- Empty time.Time fields are omitted (SDK can't parse "")
Store fix:
- store.list() snapshots under RLock then iterates without lock,
preventing the RWMutex upgrade deadlock that hung GET /v1/tasks/...
whenever the cron sweeper's put() was waiting on a Lock that
competed with reader RLocks
go test ./pkg/sdk/examples/... → 4/4 PASS:
TestHealth, TestHello, TestSignal, TestCron
61 lines
1.5 KiB
Go
61 lines
1.5 KiB
Go
// Copyright © 2026 Hanzo AI. MIT License.
|
|
|
|
package examples
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/hanzoai/tasks/pkg/sdk/client"
|
|
)
|
|
|
|
// SignalResult captures the lifecycle of a signal-driven workflow.
|
|
type SignalResult struct {
|
|
WorkflowID string
|
|
RunID string
|
|
HistoryBefore int64
|
|
HistoryAfter int64
|
|
FinalStatus int
|
|
}
|
|
|
|
// Signal drives a workflow + a signal + a terminate. The engine doesn't
|
|
// run the workflow's signal channel goroutine yet, but the history
|
|
// length must increment on each signal.
|
|
func Signal(ctx context.Context, c client.Client, _ string) (*SignalResult, error) {
|
|
run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
|
ID: "signal-" + time.Now().UTC().Format("150405.000000"),
|
|
TaskQueue: "signals",
|
|
}, "WaitingWorker")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
id, runID := run.GetID(), run.GetRunID()
|
|
before, err := c.DescribeWorkflow(ctx, id, runID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := c.SignalWorkflow(ctx, id, runID, "go", map[string]any{"step": 1}); err != nil {
|
|
return nil, err
|
|
}
|
|
after, err := c.DescribeWorkflow(ctx, id, runID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := c.TerminateWorkflow(ctx, id, runID, "test cleanup"); err != nil {
|
|
return nil, err
|
|
}
|
|
final, err := c.DescribeWorkflow(ctx, id, runID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &SignalResult{
|
|
WorkflowID: id,
|
|
RunID: runID,
|
|
HistoryBefore: before.HistoryLength,
|
|
HistoryAfter: after.HistoryLength,
|
|
FinalStatus: int(final.Status),
|
|
}, nil
|
|
}
|