hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(transport): address the engine and the SDK, so both can use a socket

Browse Source 
The engine took a PORT and the SDK an Options.HostPort, so neither end
could use a unix socket for traffic that never leaves the host. The
field name was not the only obstacle: luxfi/zap bound net.Listen("tcp",
":port") and reached a peer with net.DialTimeout("tcp", addr), so a path
was refused at the dialer with "missing port in address" and could not
be bound at all. That is fixed upstream (luxfi/zap Address + Network);
this pins the commit until it is tagged.

EmbedConfig.Address, Embedded.Address, ServeGated(addr) and
client.Options.Address each take ONE address, because a port is a detail
of one transport rather than a thing every listener has. A path binds or
dials a unix socket; anything else is host:port.

TestDialRequiresHostPort became TestDialRequiresAddress: Dial needs an
address, and asserts a socket path is refused for being absent rather
than for being portless. TestE2E_UnixSocket runs a real workflow over a
socket end to end, and TestE2E_UnixSocketRebind proves the socket is
unlinked on stop so a restart binds cleanly.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-07-28 16:27:19 -07:00
co-authored by Hanzo Dev
parent b4529923c1
commit 679ae61163
44 changed files with 323 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CONTRACT.md
+4 -2
View File
@@ -103,7 +103,7 @@ import (
)
cli, err := tasksclient.Dial(tasksclient.Options{
HostPort: os.Getenv("TASKS_ADDR"), // "tasks.hanzo.svc:9999"
Address: os.Getenv("TASKS_ADDR"), // "tasks.hanzo.svc:9999" or "/run/tasks.sock"
Namespace: tenantNamespace(orgID), // see §6
})
if err != nil {
@@ -154,7 +154,9 @@ ID — that is your idempotency token.
## 5. ZAP transport
Production: ZAP only. `tasksd` listens on TCP/9999 by default; the SDK
Production: ZAP only. `tasksd` listens on TCP/9999 by default -- or on a
unix socket when `--zap` is given a path, which is the right shape when
caller and engine share a host; the SDK
dials directly. No HTTP, no JSON envelope on the hot path, no gRPC.
Local dev: `TASKS_ADDR=""` switches the SDK to in-process mode (no
Dockerfile
+1 -1
View File
@@ -49,4 +49,4 @@ RUN apk add --no-cache ca-certificates && mkdir -p /data
COPY --from=go-build /tasksd /usr/local/bin/tasksd
EXPOSE 9999 7243
ENTRYPOINT ["tasksd"]
CMD ["--zap-port", "9999", "--http", ":7243", "--data", "/data"]
CMD ["--zap", ":9999", "--http", ":7243", "--data", "/data"]
LLM.md
+4 -2
View File
@@ -42,7 +42,9 @@ go build ./cmd/tasksd/
### Live at tasks.hanzo.ai
- **Cluster**: hanzo-k8s (do-sfo3-hanzo-k8s), namespace `hanzo`
- **Server**: `ghcr.io/hanzoai/tasks:latest` -- ZAP on 9999, HTTP on 7243
- **Server**: `ghcr.io/hanzoai/tasks:latest` -- ZAP on 9999, HTTP on 7243.
`--zap` takes an ADDRESS: `:9999` for TCP, or a path to bind a unix
socket for service-to-service traffic that never leaves the host.
- **UI**: `ghcr.io/hanzoai/tasks-ui:latest` -- port 8080
- **Database**: PostgreSQL at `sql.hanzo.svc:5432` (databases: `tasks`, `tasks_visibility`)
- **Secrets**: KMS-managed via `tasks-secrets` (POSTGRES_PASSWORD, TASKS_AUTH_SECRET, IAM_CLIENT_SECRET)
@@ -155,7 +157,7 @@ schema/tasks.zap # canonical wire schema
```
Build proof: `go build ./cmd/tasksd` → 10 MB native binary, 208 deps.
Boot proof: `tasksd --zap-port 9999 --http :7243` listens on both,
Boot proof: `tasksd --zap :9999 --http :7243` listens on both,
serves `/healthz`, `/v1/tasks/health`, `/_/tasks/*` (UI), responds to
ZAP opcodes 0x0050–0x00A5 from `pkg/sdk/client`.
cmd/tasksd/main.go
+3 -13
View File
@@ -74,7 +74,7 @@ func main() {
return
}
var (
zapPort = flag.Int("zap-port", envInt("TASKS_ZAP_PORT", 9999), "ZAP listener port")
zapAddr = flag.String("zap", envStr("TASKS_ZAP_ADDR", ":9999"), "ZAP listen address: a host:port, or a path to bind a unix socket")
httpAddr = flag.String("http", envStr("TASKS_HTTP_ADDR", ":7243"), "HTTP listen address (UI + healthz)")
dataDir = flag.String("data", envStr("TASKSD_DATA_DIR", envStr("TASKS_DATA_DIR", "./tasks-data")), "Tasks persistence directory")
ns = flag.String("namespace", envStr("TASKS_NAMESPACE", "default"), "Default namespace")
@@ -112,7 +112,7 @@ func main() {
srv, err := tasks.Embed(ctx, tasks.EmbedConfig{
DataDir: *dataDir,
MasterKey: master,
ZAPPort: *zapPort,
Address: *zapAddr,
Namespace: *ns,
Logger: logger,
JWTValidator: validator,
@@ -126,7 +126,7 @@ func main() {
os.Exit(1)
}
defer srv.Stop(context.Background())
logger.Info("zap listener", "port", srv.ZAPPort(), "service", "_tasks._tcp")
logger.Info("zap listener", "addr", srv.Address(), "service", "_tasks._tcp")
httpSrv := &http.Server{
Addr: *httpAddr,
@@ -315,13 +315,3 @@ func stringOr(a, b string) string {
}
return b
}
func envInt(k string, def int) int {
if v := os.Getenv(k); v != "" {
var n int
if _, err := fmt.Sscanf(v, "%d", &n); err == nil {
return n
}
}
return def
}
go.mod
+1 -1
View File
@@ -13,7 +13,7 @@ require (
github.com/hanzoai/sqlite v0.4.0
github.com/luxfi/consensus v1.36.3
github.com/luxfi/log v1.4.3
github.com/luxfi/zap v1.2.6
github.com/luxfi/zap v1.2.7-0.20260728232025-5bce57952415
github.com/robfig/cron/v3 v3.0.1
golang.org/x/time v0.15.0
)
go.sum
+2
View File
@@ -90,6 +90,8 @@ github.com/luxfi/warp v1.24.1 h1:9F+z6fy4sxmXp+3LlR9m3TiZ8Dfthh+s5KOeewSJL30=
github.com/luxfi/warp v1.24.1/go.mod h1:kRGQDt6EB3oRLYo71aXqnmJuCBVfND3oQ5/2miaLoyg=
github.com/luxfi/zap v1.2.6 h1:NBpbm9Gib41Oi/XAkAZKQ3hb+xCafo7JsrUjw+bKiAc=
github.com/luxfi/zap v1.2.6/go.mod h1:sTAe/AMMamoE85cVoe81+NbqHJkgvqS0LhY9ByHEmr0=
github.com/luxfi/zap v1.2.7-0.20260728232025-5bce57952415 h1:QTVlIEzR8u72MFPk79eUGm86TNGUUqtsaJCPWD0+xSs=
github.com/luxfi/zap v1.2.7-0.20260728232025-5bce57952415/go.mod h1:cAoQ1g80iSxgWOhXj6YBrWfKJi771FjPe8HdQ2eOLZw=
github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy0/jY=
github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4=
k8s/deployment.yaml
+1 -1
View File
@@ -38,7 +38,7 @@ spec:
image: ghcr.io/hanzoai/tasks:v3.3.1
imagePullPolicy: IfNotPresent
args:
- --zap-port=9999
- --zap=:9999
- --http=:7243
- --data=/data
- --namespace=$(TASKS_NAMESPACE)
pkg/sdk/client/client.go
+5 -3
View File
@@ -242,10 +242,10 @@ func Dial(opts Options) (Client, error) {
tr := opts.Transport
if tr == nil {
if opts.HostPort == "" {
return nil, errors.New("hanzo/tasks/client: Options.HostPort is required when no Transport is injected")
if opts.Address == "" {
return nil, errors.New("hanzo/tasks/client: Options.Address is required when no Transport is injected")
}
zt, err := newZAPTransport(opts.HostPort, opts.DialTimeout, id, opts.Token)
zt, err := newZAPTransport(opts.Address, opts.DialTimeout, id, opts.Token)
if err != nil {
return nil, err
}
@@ -368,6 +368,8 @@ func newZAPTransport(addr string, dialTimeout time.Duration, nodeID string, toke
var rb [4]byte
_, _ = cryptorand.Read(rb[:])
nodeID = nodeID + "-" + hex.EncodeToString(rb[:])
// The client only dials; it binds an ephemeral TCP port of its own so
// the peer can push deliveries back to it.
node := zap.NewNode(zap.NodeConfig{
NodeID: nodeID,
ServiceType: "_tasks._tcp",
pkg/sdk/client/client_test.go
+14 -2
View File
@@ -4,6 +4,8 @@ import (
"context"
"encoding/json"
"errors"
"path/filepath"
"strings"
"testing"
"time"
@@ -65,10 +67,20 @@ func (s *stubTransport) Handle(opcode uint16, fn func(from string, body []byte))
_ = fn
}
func TestDialRequiresHostPort(t *testing.T) {
// Dial needs an address. It does not need a PORT: a filesystem path is a
// valid address and dials a unix socket.
func TestDialRequiresAddress(t *testing.T) {
t.Parallel()
if _, err := Dial(Options{}); err == nil {
t.Fatal("expected Dial to reject empty Options")
t.Fatal("expected Dial to reject Options with no address")
}
// A socket path is refused for being absent, never for being portless.
_, err := Dial(Options{Address: filepath.Join(t.TempDir(), "absent.sock")})
if err == nil {
t.Fatal("expected Dial to fail against a socket nothing is serving")
}
if strings.Contains(err.Error(), "missing port") {
t.Fatalf("a socket path must not be parsed as host:port: %v", err)
}
}
pkg/sdk/client/events.go
+1 -1
View File
@@ -41,7 +41,7 @@ type Event struct {
// is canceled or the connection drops. Cancel ctx to stop.
//
// httpAddr is the daemon's HTTP listener (e.g. "127.0.0.1:7243").
// Pass empty to use http://<HostPort host>:7243 — the canonical sibling
// Pass empty to use http://<Address host>:7243 — the canonical sibling
// port shipped by tasksd alongside ZAP :9999.
func SubscribeEvents(ctx context.Context, httpAddr string) (<-chan Event, error) {
if httpAddr == "" {
pkg/sdk/client/options.go
+6 -3
View File
@@ -20,9 +20,12 @@ import (
// Options configures Dial. Fields mirror the upstream Temporal Options
// surface so callers can swap import paths without changing arguments.
type Options struct {
// HostPort is the "host:port" address of the Hanzo Tasks frontend
// exposing the ZAP transport (default port 9999).
HostPort string
// Address is where the Hanzo Tasks frontend exposes its ZAP
// transport. A filesystem path dials a unix socket — the right shape
// for service-to-service traffic that never leaves the host, and one
// that needs no port. Anything else is a host:port TCP address.
// Required unless Transport is injected.
Address string
// Namespace scopes every RPC issued by the returned Client. Empty
// string is rewritten to "default" at Dial time.
pkg/sdk/client/schedule.go
+10 -10
View File
@@ -146,9 +146,9 @@ type updateScheduleRequest struct {
}
type triggerScheduleRequest struct {
Namespace string `json:"namespace"`
ScheduleID string `json:"schedule_id"`
OverlapPolicy string `json:"overlap_policy,omitempty"`
Namespace string `json:"namespace"`
ScheduleID string `json:"schedule_id"`
OverlapPolicy string `json:"overlap_policy,omitempty"`
}
type describeScheduleRequest struct {
@@ -158,17 +158,17 @@ type describeScheduleRequest struct {
// DescribeScheduleResponse is the v1 wire response for DescribeSchedule.
type DescribeScheduleResponse struct {
Schedule Schedule `json:"schedule"`
Info ScheduleInfo `json:"info"`
Schedule Schedule `json:"schedule"`
Info ScheduleInfo `json:"info"`
}
// ScheduleInfo carries runtime state for a schedule.
type ScheduleInfo struct {
ActionCount int64 `json:"action_count"`
MissedCatchupCount int64 `json:"missed_catchup_count"`
OverlapSkipped int64 `json:"overlap_skipped"`
BufferDropped int64 `json:"buffer_dropped"`
BufferSize int64 `json:"buffer_size"`
ActionCount int64 `json:"action_count"`
MissedCatchupCount int64 `json:"missed_catchup_count"`
OverlapSkipped int64 `json:"overlap_skipped"`
BufferDropped int64 `json:"buffer_dropped"`
BufferSize int64 `json:"buffer_size"`
RunningWorkflows []ScheduleListEntry `json:"running_workflows,omitempty"`
}
pkg/sdk/client/transport.go
+35 -35
View File
@@ -15,12 +15,12 @@ import "context"
//
// Server-push wire model (replaces long-poll, 2026-04-28):
//
// Workers Subscribe once per (namespace, taskQueue, kind); the server
// pushes Deliver{Workflow,Activity}Task as work arrives, and pushes
// DeliverActivityResult to the workflow's worker when an activity
// completes. There is no polling. The authoritative wire shapes live
// in pkg/tasks/dispatch.go (workflowTaskDeliveryJSON,
// activityTaskDeliveryJSON, activityResultDeliveryJSON).
// Workers Subscribe once per (namespace, taskQueue, kind); the server
// pushes Deliver{Workflow,Activity}Task as work arrives, and pushes
// DeliverActivityResult to the workflow's worker when an activity
// completes. There is no polling. The authoritative wire shapes live
// in pkg/tasks/dispatch.go (workflowTaskDeliveryJSON,
// activityTaskDeliveryJSON, activityResultDeliveryJSON).
//
// Append-only across releases; never reuse a value. 0x00A0 and 0x00A1
// were repurposed from Poll{Workflow,Activity}Task to
@@ -72,12 +72,12 @@ const (
FieldWorkerBuildID = 32
// WorkflowTask response fields.
FieldTaskToken = 0
FieldWorkflowID = 8
FieldRunID = 16
FieldWorkflowTypeName = 24
FieldHistoryBytes = 32
FieldNextPageToken = 40
FieldTaskToken = 0
FieldWorkflowID = 8
FieldRunID = 16
FieldWorkflowTypeName = 24
FieldHistoryBytes = 32
FieldNextPageToken = 40
// ActivityTask response fields.
FieldActivityID = 8
@@ -90,10 +90,10 @@ const (
FieldActivityRunID = 64
// Respond request fields.
FieldCommandsBytes = 8 // workflow task completed
FieldResultBytes = 8 // activity task completed
FieldFailureBytes = 8 // activity task failed
FieldDetailsBytes = 8 // heartbeat
FieldCommandsBytes = 8 // workflow task completed
FieldResultBytes = 8 // activity task completed
FieldFailureBytes = 8 // activity task failed
FieldDetailsBytes = 8 // heartbeat
// Response fields.
FieldRespStatus = 0
@@ -107,11 +107,11 @@ const (
//
// Wire model (server-push, 2026-04-28):
//
// The worker subscribes once per (namespace, taskQueue, kind) and
// registers OnWorkflowTask / OnActivityTask / OnActivityResult
// callbacks. The server pushes work as it arrives. The worker
// responds via Respond{Workflow,Activity}Task{Completed,Failed} and
// RecordActivityTaskHeartbeat — same as before. There is no polling.
// The worker subscribes once per (namespace, taskQueue, kind) and
// registers OnWorkflowTask / OnActivityTask / OnActivityResult
// callbacks. The server pushes work as it arrives. The worker
// responds via Respond{Workflow,Activity}Task{Completed,Failed} and
// RecordActivityTaskHeartbeat — same as before. There is no polling.
type WorkerTransport interface {
// Close tears down the underlying node. Safe to call more than
// once.
@@ -203,11 +203,11 @@ type StartChildWorkflowResponse struct {
// PollWorkflowTaskRequest mirrors schema/tasks.zap:PollWorkflowTaskRequest.
type PollWorkflowTaskRequest struct {
Namespace string
TaskQueueName string
TaskQueueKind int8
Identity string
WorkerBuildID string
Namespace string
TaskQueueName string
TaskQueueKind int8
Identity string
WorkerBuildID string
}
// PollActivityTaskRequest mirrors schema/tasks.zap:PollActivityTaskRequest.
@@ -232,15 +232,15 @@ type WorkflowTask struct {
// encoded Payloads; the worker's activity dispatcher decodes it into
// the registered function's arg types.
type ActivityTask struct {
TaskToken []byte
WorkflowID string
RunID string
ActivityID string
ActivityTypeName string
Input []byte
ScheduledTimeMs int64
StartToCloseTimeoutMs int64
HeartbeatTimeoutMs int64
TaskToken []byte
WorkflowID string
RunID string
ActivityID string
ActivityTypeName string
Input []byte
ScheduledTimeMs int64
StartToCloseTimeoutMs int64
HeartbeatTimeoutMs int64
}
// RespondWorkflowTaskCompletedRequest carries a serialised command list.
pkg/sdk/client/workflow.go
+26 -26
View File
@@ -46,29 +46,29 @@ type WorkflowRunGetOptions struct {
// for the v1 JSON wire. Exported so callers of DescribeWorkflow /
// ListWorkflows can inspect the result without pulling upstream types.
type WorkflowExecutionInfo struct {
WorkflowID string `json:"workflow_id"`
RunID string `json:"run_id"`
WorkflowType string `json:"workflow_type"`
StartTime time.Time `json:"start_time,omitempty"`
CloseTime time.Time `json:"close_time,omitempty"`
Status WorkflowStatus `json:"status"`
HistoryLength int64 `json:"history_length"`
TaskQueue string `json:"task_queue,omitempty"`
Memo map[string]any `json:"memo,omitempty"`
WorkflowID string `json:"workflow_id"`
RunID string `json:"run_id"`
WorkflowType string `json:"workflow_type"`
StartTime time.Time `json:"start_time,omitempty"`
CloseTime time.Time `json:"close_time,omitempty"`
Status WorkflowStatus `json:"status"`
HistoryLength int64 `json:"history_length"`
TaskQueue string `json:"task_queue,omitempty"`
Memo map[string]any `json:"memo,omitempty"`
}
// WorkflowStatus mirrors the `status` Int8 in schema/tasks.zap.
type WorkflowStatus int8
const (
WorkflowStatusUnspecified WorkflowStatus = 0
WorkflowStatusRunning WorkflowStatus = 1
WorkflowStatusCompleted WorkflowStatus = 2
WorkflowStatusFailed WorkflowStatus = 3
WorkflowStatusCanceled WorkflowStatus = 4
WorkflowStatusTerminated WorkflowStatus = 5
WorkflowStatusContinuedAsNew WorkflowStatus = 6
WorkflowStatusTimedOut WorkflowStatus = 7
WorkflowStatusUnspecified WorkflowStatus = 0
WorkflowStatusRunning WorkflowStatus = 1
WorkflowStatusCompleted WorkflowStatus = 2
WorkflowStatusFailed WorkflowStatus = 3
WorkflowStatusCanceled WorkflowStatus = 4
WorkflowStatusTerminated WorkflowStatus = 5
WorkflowStatusContinuedAsNew WorkflowStatus = 6
WorkflowStatusTimedOut WorkflowStatus = 7
)
// ListWorkflowsResponse is returned by ListWorkflows.
@@ -80,16 +80,16 @@ type ListWorkflowsResponse struct {
// startWorkflowRequest is the v1 JSON shape mapped onto schema/tasks.zap
// StartWorkflowRequest. Native ZAP serde replaces JSON in a follow-up.
type startWorkflowRequest struct {
Namespace string `json:"namespace"`
WorkflowID string `json:"workflow_id"`
WorkflowType string `json:"workflow_type"`
TaskQueue string `json:"task_queue"`
Input []any `json:"input,omitempty"`
RetryPolicy *retryPolicy `json:"retry_policy,omitempty"`
Timeouts timeouts `json:"timeouts,omitempty"`
Namespace string `json:"namespace"`
WorkflowID string `json:"workflow_id"`
WorkflowType string `json:"workflow_type"`
TaskQueue string `json:"task_queue"`
Input []any `json:"input,omitempty"`
RetryPolicy *retryPolicy `json:"retry_policy,omitempty"`
Timeouts timeouts `json:"timeouts,omitempty"`
Memo map[string]any `json:"memo,omitempty"`
CronSchedule string `json:"cron_schedule,omitempty"`
Identity string `json:"identity,omitempty"`
CronSchedule string `json:"cron_schedule,omitempty"`
Identity string `json:"identity,omitempty"`
}
type retryPolicy struct {
pkg/sdk/examples/embed.go
+8 -8
View File
@@ -14,12 +14,12 @@ import (
// EmbedResult captures the lifecycle of an in-process daemon dial.
type EmbedResult struct {
Port int
HealthService string
HealthStatus string
StartedID string
StartedRunID string
ListedCount int
Port int
HealthService string
HealthStatus string
StartedID string
StartedRunID string
ListedCount int
}
// Embed boots tasks.Embed in-process on an ephemeral port, dials it via
@@ -38,7 +38,7 @@ func Embed(ctx context.Context) (*EmbedResult, error) {
_ = l.Close()
srv, err := tasks.Embed(ctx, tasks.EmbedConfig{
ZAPPort: port,
Address: fmt.Sprintf(":%d", port),
Namespace: "default",
})
if err != nil {
@@ -47,7 +47,7 @@ func Embed(ctx context.Context) (*EmbedResult, error) {
defer func() { _ = srv.Stop(context.Background()) }()
c, err := client.Dial(client.Options{
HostPort: fmt.Sprintf("127.0.0.1:%d", port),
Address: fmt.Sprintf("127.0.0.1:%d", port),
Namespace: "default",
DialTimeout: 3 * time.Second,
CallTimeout: 3 * time.Second,
pkg/sdk/examples/examples_test.go
+2 -2
View File
@@ -24,14 +24,14 @@ func boot(t *testing.T) (client.Client, func()) {
t.Helper()
port := freePort(t)
srv, err := tasks.Embed(context.Background(), tasks.EmbedConfig{
ZAPPort: port,
Address: fmt.Sprintf(":%d", port),
Namespace: "default",
})
if err != nil {
t.Fatalf("tasks.Embed: %v", err)
}
c, err := client.Dial(client.Options{
HostPort: fmt.Sprintf("127.0.0.1:%d", port),
Address: fmt.Sprintf("127.0.0.1:%d", port),
Namespace: "default",
DialTimeout: 3 * time.Second,
CallTimeout: 3 * time.Second,
pkg/sdk/examples/realtime.go
+6 -6
View File
@@ -16,10 +16,10 @@ import (
// RealtimeResult captures what arrived on the SSE stream during the
// demo window: every workflow.* event for the workflows we started.
type RealtimeResult struct {
StartedCount int
EventsCaught int
StartEvents int
CancelEvents int
StartedCount int
EventsCaught int
StartEvents int
CancelEvents int
}
// Realtime boots an embedded daemon, opens an SSE subscription via
@@ -44,7 +44,7 @@ func Realtime(parent context.Context) (*RealtimeResult, error) {
bootCtx, bootCancel := context.WithCancel(parent)
defer bootCancel()
srv, err := tasks.Embed(bootCtx, tasks.EmbedConfig{
ZAPPort: port,
Address: fmt.Sprintf(":%d", port),
Namespace: "default",
})
if err != nil {
@@ -74,7 +74,7 @@ func Realtime(parent context.Context) (*RealtimeResult, error) {
// Dial via ZAP and fire a couple state changes.
c, err := client.Dial(client.Options{
HostPort: fmt.Sprintf("127.0.0.1:%d", port),
Address: fmt.Sprintf("127.0.0.1:%d", port),
Namespace: "default",
DialTimeout: 1 * time.Second,
CallTimeout: 1 * time.Second,
pkg/sdk/inproc/inproc.go
+4 -4
View File
@@ -122,7 +122,7 @@ func (t *Transport) Close() error {
// both an explicit Transport and a Dispatcher is ambiguous and is
// rejected.
//
// opts.HostPort is ignored for inproc (no dial happens). All other
// opts.Address is ignored for inproc (no dial happens). All other
// fields (Namespace, Identity, CallTimeout, ...) flow through to
// [client.Dial] unchanged.
func NewClient(dispatcher Dispatcher, opts client.Options) (client.Client, error) {
@@ -130,11 +130,11 @@ func NewClient(dispatcher Dispatcher, opts client.Options) (client.Client, error
return nil, errors.New("hanzo/tasks/inproc: opts.Transport must be nil — use either inproc or a Transport, not both")
}
opts.Transport = NewTransport(dispatcher)
// HostPort is required by Dial when no Transport is set; we set
// Address is required by Dial when no Transport is set; we set
// the Transport above, so the value is ignored downstream. Pass a
// sentinel so logs are unambiguous.
if opts.HostPort == "" {
opts.HostPort = "inproc://frontend"
if opts.Address == "" {
opts.Address = "inproc://frontend"
}
return client.Dial(opts)
}
pkg/sdk/inproc/inproc_test.go
+4 -4
View File
@@ -136,7 +136,7 @@ func TestNewClient_RejectsTransportConflict(t *testing.T) {
}
}
func TestNewClient_DefaultsHostPortSentinel(t *testing.T) {
func TestNewClient_DefaultsAddressSentinel(t *testing.T) {
d := &fakeDispatcher{}
c, err := NewClient(d, client.Options{Namespace: "default"})
if err != nil {
@@ -153,9 +153,9 @@ func TestNewClient_DefaultsHostPortSentinel(t *testing.T) {
t.Skip("fake dispatcher echo accidentally satisfied health decode")
}
// We only assert that the failure is from response decode, not
// from HostPort dial — the latter would be "Options.HostPort is
// from Address dial — the latter would be "Options.Address is
// required" before any RPC is issued.
if got := err.Error(); got == "hanzo/tasks/client: Options.HostPort is required when no Transport is injected" {
t.Fatalf("inproc.NewClient should not require HostPort, got %v", err)
if got := err.Error(); got == "hanzo/tasks/client: Options.Address is required when no Transport is injected" {
t.Fatalf("inproc.NewClient should not require Address, got %v", err)
}
}
pkg/sdk/worker/env.go
-1
View File
@@ -679,4 +679,3 @@ func (e *workerEnv) PendingUpserts() []map[string]any {
e.pendingUpserts = nil
return out
}
pkg/sdk/workflow/child.go
+3 -3
View File
@@ -62,9 +62,9 @@ func (f *childFuture) Get(ctx Context, valPtr any) error {
return f.result.Get(ctx, valPtr)
}
func (f *childFuture) IsReady() bool { return f.result.IsReady() }
func (f *childFuture) ReadyCh() <-chan struct{} { return f.result.ReadyCh() }
func (f *childFuture) GetChildWorkflowExecution() Future { return f.execution }
func (f *childFuture) IsReady() bool { return f.result.IsReady() }
func (f *childFuture) ReadyCh() <-chan struct{} { return f.result.ReadyCh() }
func (f *childFuture) GetChildWorkflowExecution() Future { return f.execution }
// ExecuteChildWorkflow is the workflow-code entry point. It delegates
// to CoroutineEnv.ExecuteChildWorkflow, which in workerEnv forwards
pkg/sdk/workflow/env.go
+1 -1
View File
@@ -3,7 +3,7 @@
// its helpers (ExecuteActivity, GetSignalChannel, NewSelector, …)
// to express durable orchestration.
//
// Layering
// # Layering
//
// The package is DELIBERATELY THIN. All runtime behaviour —
// deterministic time, activity dispatch, signal queues, selector
pkg/sdk/workflow/selector.go
+6 -6
View File
@@ -24,12 +24,12 @@ func NewSelector(ctx Context) Selector {
}
type selectorCase struct {
kind int // 0 = future, 1 = receive, 2 = default
future Future
futCB func(Future)
ch ReceiveChannel
chCB func(ReceiveChannel, bool)
defCB func()
kind int // 0 = future, 1 = receive, 2 = default
future Future
futCB func(Future)
ch ReceiveChannel
chCB func(ReceiveChannel, bool)
defCB func()
}
const (
pkg/sdk/workflow/stub.go
+1 -1
View File
@@ -650,7 +650,7 @@ func (stubEnv) MutableSideEffect(_ string, fn func(ctx Context) any, _ func(a, b
}
return EncodePayload(fn(ctx))
}
func (stubEnv) MetricsHandler() MetricsHandler { return nil }
func (stubEnv) MetricsHandler() MetricsHandler { return nil }
func (stubEnv) UpsertSearchAttributes(map[string]any) error { return nil }
// activityName derives the activity's registered name. For function
pkg/sdk/workflow/timer.go
+4 -4
View File
@@ -48,10 +48,10 @@ func NewTimer(ctx Context, d time.Duration) Future {
// time.Timer. Exported so the worker package (a sibling) can reuse
// the same implementation without reaching into package internals.
//
// - d <= 0 settles the Future with (nil, nil) immediately.
// - Otherwise, a per-timer goroutine parks on the timer's channel
// alongside the supplied doneCh. If doneCh closes first, the Future
// settles with errFn() — typically the scope's cancel error.
// - d <= 0 settles the Future with (nil, nil) immediately.
// - Otherwise, a per-timer goroutine parks on the timer's channel
// alongside the supplied doneCh. If doneCh closes first, the Future
// settles with errFn() — typically the scope's cancel error.
//
// Cancellation semantics: a closed doneCh always wins over a fired
// timer in the rare race where both are observable simultaneously.
pkg/sdk/workflow/version_test.go
+6 -4
View File
@@ -199,11 +199,13 @@ type fakeMetrics struct {
counters map[string]int64
}
func newFake() *fakeMetrics { return &fakeMetrics{counters: make(map[string]int64)} }
func newFake() *fakeMetrics { return &fakeMetrics{counters: make(map[string]int64)} }
func (f *fakeMetrics) WithTags(map[string]string) workflow.MetricsHandler { return f }
func (f *fakeMetrics) Counter(name string) workflow.MetricsCounter { return &fakeCounter{f: f, name: name} }
func (f *fakeMetrics) Gauge(string) workflow.MetricsGauge { return &fakeGauge{} }
func (f *fakeMetrics) Timer(string) workflow.MetricsTimer { return &fakeTimer{} }
func (f *fakeMetrics) Counter(name string) workflow.MetricsCounter {
return &fakeCounter{f: f, name: name}
}
func (f *fakeMetrics) Gauge(string) workflow.MetricsGauge { return &fakeGauge{} }
func (f *fakeMetrics) Timer(string) workflow.MetricsTimer { return &fakeTimer{} }
type fakeCounter struct {
f *fakeMetrics
pkg/tasks/activities.go
+4 -4
View File
@@ -531,10 +531,10 @@ func (e *engine) CreateVersion(ns, name, buildId, description, compute, image st
// DeploymentVersionPatch is metadata-only — buildId and CreateTime are
// immutable.
type DeploymentVersionPatch struct {
Description *string `json:"description,omitempty"`
Compute *string `json:"compute,omitempty"`
Image *string `json:"image,omitempty"`
Env map[string]string `json:"env,omitempty"`
Description *string `json:"description,omitempty"`
Compute *string `json:"compute,omitempty"`
Image *string `json:"image,omitempty"`
Env map[string]string `json:"env,omitempty"`
}
// UpdateVersion patches metadata on a deployment version. buildId is
pkg/tasks/activities_test.go
+1 -1
View File
@@ -158,7 +158,7 @@ func TestHTTP_Activities_FailFlow(t *testing.T) {
en, h := httpFixture(t)
a, _ := en.StartActivity("default", "a-f", "", TypeRef{Name: "T"}, "q", nil, nil, "", "", "", "", "", "")
run := a.Execution.RunId
code, _ := httpDo(t, h, http.MethodPost, "/v1/tasks/namespaces/default/activities/a-f/"+run+"/fail", map[string]any{"cause": "boom"}, )
code, _ := httpDo(t, h, http.MethodPost, "/v1/tasks/namespaces/default/activities/a-f/"+run+"/fail", map[string]any{"cause": "boom"})
if code != 200 {
t.Fatalf("fail code=%d", code)
}
pkg/tasks/cancel_for_org_test.go
+2 -2
View File
@@ -12,7 +12,7 @@ import (
// in the caller's shard, is rejected once the job is terminal, and never reaches
// across the org boundary — the same tenancy contract ActivitiesForOrg reads under.
func TestCancelActivityForOrg(t *testing.T) {
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0})
emb, err := Embed(context.Background(), EmbedConfig{Address: ""})
if err != nil {
t.Fatalf("embed: %v", err)
}
@@ -59,7 +59,7 @@ func TestCancelActivityForOrg(t *testing.T) {
// activity, not just the hash-ordered first 100 ActivitiesForOrg caps at — the
// truncation that hid live jobs and dropped online workers on a busy org.
func TestActivitiesPageForOrgWalksPastFirstPage(t *testing.T) {
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0})
emb, err := Embed(context.Background(), EmbedConfig{Address: ""})
if err != nil {
t.Fatalf("embed: %v", err)
}
pkg/tasks/client_test.go
+2 -1
View File
@@ -2,6 +2,7 @@ package tasks
import (
"context"
"fmt"
"net"
"strconv"
"sync"
@@ -29,7 +30,7 @@ func bootServer(t *testing.T) (*Embedded, string) {
t.Helper()
ctx := context.Background()
port := pickPort(t)
emb, err := Embed(ctx, EmbedConfig{ZAPPort: port})
emb, err := Embed(ctx, EmbedConfig{Address: fmt.Sprintf(":%d", port)})
if err != nil {
t.Fatalf("embed: %v", err)
}
pkg/tasks/cluster_test.go
+2 -2
View File
@@ -13,7 +13,7 @@ import (
)
func TestCluster_StatusAndHealth(t *testing.T) {
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0, NodeID: "n1"})
emb, err := Embed(context.Background(), EmbedConfig{Address: "", NodeID: "n1"})
if err != nil {
t.Fatalf("embed: %v", err)
}
@@ -53,7 +53,7 @@ func TestCluster_StatusAndHealth(t *testing.T) {
}
func TestCluster_MigrateEndpoint(t *testing.T) {
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0, NodeID: "n1"})
emb, err := Embed(context.Background(), EmbedConfig{Address: "", NodeID: "n1"})
if err != nil {
t.Fatalf("embed: %v", err)
}
pkg/tasks/dispatch_test.go
-1
View File
@@ -175,7 +175,6 @@ func TestDispatcher_QueuesUntilSubscribe(t *testing.T) {
}
}
// TestDispatcher_PrunesDeadPeerOnSendFailure proves the self-healing delivery
// path: a leaked subscription whose worker has disconnected (Send fails, since
// luxfi/zap exposes no disconnect hook) is pruned on the failed push and the
pkg/tasks/durable_e2e_test.go
+5 -5
View File
@@ -25,7 +25,7 @@ import (
func dialLocal(t *testing.T, port int) client.Client {
t.Helper()
cli, err := client.Dial(client.Options{
HostPort: fmt.Sprintf("127.0.0.1:%d", port),
Address: fmt.Sprintf("127.0.0.1:%d", port),
Namespace: "default",
DialTimeout: 5 * time.Second,
})
@@ -107,7 +107,7 @@ func TestDurable_ActivityRetryBackoff(t *testing.T) {
flakyMu.Unlock()
port := freePort(t)
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{ZAPPort: port})
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: fmt.Sprintf(":%d", port)})
if err != nil {
t.Fatalf("embed: %v", err)
}
@@ -191,7 +191,7 @@ func TestDurable_ActivityRetryExhaustion_FailsWorkflow(t *testing.T) {
exhaustAttempts.Store(0)
port := freePort(t)
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{ZAPPort: port})
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: fmt.Sprintf(":%d", port)})
if err != nil {
t.Fatalf("embed: %v", err)
}
@@ -268,7 +268,7 @@ func TestDurable_CrashRecovery_ReachesCompleted(t *testing.T) {
// Embed A + worker A: drive the activity in-flight, then simulate a crash.
portA := freePort(t)
embA, err := tasks.Embed(ctx, tasks.EmbedConfig{ZAPPort: portA, DataDir: dir})
embA, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: fmt.Sprintf(":%d", portA), DataDir: dir})
if err != nil {
t.Fatalf("embed A: %v", err)
}
@@ -304,7 +304,7 @@ func TestDurable_CrashRecovery_ReachesCompleted(t *testing.T) {
// runs the now-deterministic activity to completion.
recovered.Store(true)
portB := freePort(t)
embB, err := tasks.Embed(ctx, tasks.EmbedConfig{ZAPPort: portB, DataDir: dir})
embB, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: fmt.Sprintf(":%d", portB), DataDir: dir})
if err != nil {
t.Fatalf("embed B: %v", err)
}
pkg/tasks/e2e_test.go
+3 -3
View File
@@ -68,14 +68,14 @@ func TestE2E_WorkflowExecutesActivityAndCompletes(t *testing.T) {
defer cancel()
port := freePort(t)
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{ZAPPort: port})
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: fmt.Sprintf(":%d", port)})
if err != nil {
t.Fatalf("embed: %v", err)
}
defer emb.Stop(ctx)
cli, err := client.Dial(client.Options{
HostPort: fmt.Sprintf("127.0.0.1:%d", port),
Address: fmt.Sprintf("127.0.0.1:%d", port),
Namespace: "default",
DialTimeout: 5 * time.Second,
})
@@ -149,7 +149,7 @@ func TestE2E_WorkflowExecutesActivityAndCompletes(t *testing.T) {
// the React UI does, with no worker registered (so the workflow stays
// in RUNNING until terminated).
func TestE2E_HTTPLifecycle(t *testing.T) {
emb, err := tasks.Embed(context.Background(), tasks.EmbedConfig{ZAPPort: 0})
emb, err := tasks.Embed(context.Background(), tasks.EmbedConfig{Address: ""})
if err != nil {
t.Fatalf("embed: %v", err)
}
pkg/tasks/embed.go
+15 -11
View File
@@ -34,8 +34,12 @@ var base64Std = base64.StdEncoding
// EmbedConfig configures the in-process Tasks server.
type EmbedConfig struct {
DataDir string // "" → "./tasks-data" (reserved; memdb today)
ZAPPort int // 0 → 9999
DataDir string // "" → "./tasks-data" (reserved; memdb today)
// Address is where the ZAP listener binds. A filesystem path binds a
// unix socket — the right shape for service-to-service traffic that
// never leaves the host, and one that needs no port. Anything else is
// a host:port TCP address; "" → ":9999".
Address string
Namespace string // "" → "default"
Logger *slog.Logger // nil → slog.Default()
// MasterKey is the 32-byte root key every shard is encrypted under.
@@ -125,7 +129,7 @@ func Embed(ctx context.Context, cfg EmbedConfig) (*Embedded, error) {
node := zap.NewNode(zap.NodeConfig{
NodeID: "tasks-embed",
ServiceType: "_tasks._tcp",
Port: cfg.ZAPPort,
Address: cfg.Address,
Logger: cfg.Logger,
NoDiscovery: true,
})
@@ -193,12 +197,12 @@ func wireSend(opcode uint16, body []byte) (*zap.Message, error) {
return zap.Parse(frame)
}
// ZAPPort returns the bound ZAP port of the loopback listener.
func (e *Embedded) ZAPPort() int {
// Address returns where the loopback ZAP listener is bound.
func (e *Embedded) Address() string {
if e == nil {
return 0
return ""
}
return e.cfg.ZAPPort
return e.cfg.Address
}
// View is an org-scoped, in-process control-plane handle on the embedded
@@ -286,7 +290,7 @@ func (e *Embedded) nodeSnapshot() []*zap.Node {
return out
}
// ServeGated starts a SECOND ZAP listener on 0.0.0.0:port that exposes the SAME
// ServeGated starts a SECOND ZAP listener at addr that exposes the SAME
// engine to out-of-process callers under mandatory identity gating: every request
// must carry an auth_token that validates against validator (RequireIdentity), and
// its engine view is org-scoped to the token owner (CONTRACT §6). The loopback
@@ -296,7 +300,7 @@ func (e *Embedded) nodeSnapshot() []*zap.Node {
// delivery. A nil validator is refused: exposing durable execution ungated across
// the cluster is never correct. Call once, after Embed; Stop tears down every
// listener.
func (e *Embedded) ServeGated(ctx context.Context, port int, validator *auth.Validator) error {
func (e *Embedded) ServeGated(ctx context.Context, addr string, validator *auth.Validator) error {
if e == nil {
return errors.New("tasks: ServeGated on nil Embedded")
}
@@ -306,7 +310,7 @@ func (e *Embedded) ServeGated(ctx context.Context, port int, validator *auth.Val
n := zap.NewNode(zap.NodeConfig{
NodeID: "tasks-embed-gated",
ServiceType: "_tasks._tcp",
Port: port,
Address: addr,
Logger: e.cfg.Logger,
NoDiscovery: true,
})
@@ -314,7 +318,7 @@ func (e *Embedded) ServeGated(ctx context.Context, port int, validator *auth.Val
n.Handle(op, h)
}
if err := n.Start(); err != nil {
return fmt.Errorf("tasks: gated zap start on :%d: %w", port, err)
return fmt.Errorf("tasks: gated zap start on %s: %w", addr, err)
}
e.nodesMu.Lock()
e.nodes = append(e.nodes, n)
pkg/tasks/engine_extras.go
+3 -3
View File
@@ -179,10 +179,10 @@ func (e *engine) DeleteDeploymentVersion(ns, name, buildId string) (*Deployment,
// "disabled" state from `enabled:false`.
func (e *engine) QueryArchival(_, _, _ string) (map[string]any, error) {
return map[string]any{
"enabled": false,
"executions": []any{},
"enabled": false,
"executions": []any{},
"nextPageToken": "",
"error": "archival disabled",
"error": "archival disabled",
}, nil
}
pkg/tasks/events.go
+5 -5
View File
@@ -21,7 +21,7 @@ import (
)
type Event struct {
Kind string `json:"kind"` // workflow.started|workflow.canceled|workflow.terminated|workflow.signaled|schedule.created|schedule.paused|schedule.resumed|schedule.deleted|namespace.registered|batch.started
Kind string `json:"kind"` // workflow.started|workflow.canceled|workflow.terminated|workflow.signaled|schedule.created|schedule.paused|schedule.resumed|schedule.deleted|namespace.registered|batch.started
OrgID string `json:"org_id,omitempty"` // tenant scope; "" = unscoped (embedded/dev)
Namespace string `json:"namespace,omitempty"`
WorkflowID string `json:"workflow_id,omitempty"`
@@ -37,10 +37,10 @@ type Event struct {
// rather than blocking the publisher. Designed for single-process tasks;
// distributed fan-out lands when persistence does.
type broker struct {
mu sync.RWMutex
nextID atomic.Uint64
subs map[uint64]chan Event
bufSize int
mu sync.RWMutex
nextID atomic.Uint64
subs map[uint64]chan Event
bufSize int
}
func newBroker() *broker {
pkg/tasks/gated_e2e_test.go
+5 -5
View File
@@ -92,25 +92,25 @@ func TestE2E_ServeGated_DurableWorkflow_RequiresIdentity(t *testing.T) {
js := gatedJWKS(t, key, "kid-1")
validator := auth.NewValidator(auth.JWTConfig{JWKSURL: js.URL, Issuer: "https://hanzo.id", TTL: time.Minute})
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{ZAPPort: freePort(t)})
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: fmt.Sprintf(":%d", freePort(t))})
if err != nil {
t.Fatalf("embed: %v", err)
}
defer emb.Stop(ctx)
// ServeGated refuses to expose the engine without a validator.
if err := emb.ServeGated(ctx, freePort(t), nil); err == nil {
if err := emb.ServeGated(ctx, fmt.Sprintf(":%d", freePort(t)), nil); err == nil {
t.Fatal("ServeGated accepted a nil validator; want refusal")
}
gatedPort := freePort(t)
if err := emb.ServeGated(ctx, gatedPort, validator); err != nil {
if err := emb.ServeGated(ctx, fmt.Sprintf(":%d", gatedPort), validator); err != nil {
t.Fatalf("serve gated: %v", err)
}
gatedAddr := fmt.Sprintf("127.0.0.1:%d", gatedPort)
// A caller WITHOUT a token is refused durable execution on the gated listener.
anon, err := client.Dial(client.Options{HostPort: gatedAddr, Namespace: "default", DialTimeout: 5 * time.Second})
anon, err := client.Dial(client.Options{Address: gatedAddr, Namespace: "default", DialTimeout: 5 * time.Second})
if err != nil {
t.Fatalf("dial anon: %v", err)
}
@@ -122,7 +122,7 @@ func TestE2E_ServeGated_DurableWorkflow_RequiresIdentity(t *testing.T) {
// A caller WITH a valid token runs a real durable workflow, org-scoped to owner.
tok := gatedSign(t, key, "kid-1", "https://hanzo.id", "acme")
cli, err := client.Dial(client.Options{
HostPort: gatedAddr,
Address: gatedAddr,
Namespace: "default",
DialTimeout: 5 * time.Second,
Token: client.StaticToken(tok),
pkg/tasks/handshake_test.go
+4 -4
View File
@@ -211,7 +211,7 @@ func TestQueryPush_WorkerTimeout(t *testing.T) {
func adminFixture(t *testing.T) (*Embedded, *httptest.Server) {
t.Helper()
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0})
emb, err := Embed(context.Background(), EmbedConfig{Address: ""})
if err != nil {
t.Fatalf("embed: %v", err)
}
@@ -247,9 +247,9 @@ func httpPost(t *testing.T, url, body string) (int, []byte) {
func TestAdmin_SearchAttributes(t *testing.T) {
_, srv := adminFixture(t)
cases := []struct {
name string
body string
code int
name string
body string
code int
}{
{"ok", `{"name":"OrderId","type":"Keyword"}`, 200},
{"duplicate", `{"name":"OrderId","type":"Keyword"}`, 409},
pkg/tasks/migration/migration.go
+4 -4
View File
@@ -31,12 +31,12 @@ import (
type Status string
const (
StatusPending Status = "pending"
StatusLocking Status = "locking"
StatusCopying Status = "copying"
StatusPending Status = "pending"
StatusLocking Status = "locking"
StatusCopying Status = "copying"
StatusReplaying Status = "replaying"
StatusReleased Status = "released"
StatusFailed Status = "failed"
StatusFailed Status = "failed"
)
// Job is a single migration unit.
pkg/tasks/replication/local.go
+6 -6
View File
@@ -12,12 +12,12 @@ import (
// subscribers — the same code path the quasar driver uses on accept,
// so callers cannot tell the difference.
type LocalReplicator struct {
mu sync.RWMutex
handlers []Handler
accepted uint64
rejected uint64
frameSeqMu sync.Mutex
nextSeq uint64
mu sync.RWMutex
handlers []Handler
accepted uint64
rejected uint64
frameSeqMu sync.Mutex
nextSeq uint64
}
// NewLocal returns a single-node Replicator.
pkg/tasks/socket_e2e_test.go
+95
View File
@@ -0,0 +1,95 @@
// Copyright © 2026 Hanzo AI. MIT License.
package tasks_test
import (
"context"
"os"
"path/filepath"
"testing"
"time"
"github.com/hanzoai/tasks/pkg/sdk/client"
"github.com/hanzoai/tasks/pkg/tasks"
)
// The engine serves, and the SDK reaches it, over a unix socket — the
// shape internal service-to-service traffic should take when both ends
// share a host. No port is involved anywhere.
func TestE2E_UnixSocket(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
sock := filepath.Join(t.TempDir(), "tasks.sock")
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: sock})
if err != nil {
t.Fatalf("Embed on %s: %v", sock, err)
}
defer emb.Stop(context.Background())
fi, err := os.Stat(sock)
if err != nil {
t.Fatalf("engine did not bind the socket: %v", err)
}
if fi.Mode()&os.ModeSocket == 0 {
t.Fatalf("%s is not a socket", sock)
}
if emb.Address() != sock {
t.Fatalf("Address() = %q want %q", emb.Address(), sock)
}
c, err := client.Dial(client.Options{
Address: sock,
Namespace: "default",
DialTimeout: 5 * time.Second,
CallTimeout: 5 * time.Second,
})
if err != nil {
t.Fatalf("Dial %s: %v", sock, err)
}
defer c.Close()
health, err := c.CheckHealth(ctx, &client.CheckHealthRequest{})
if err != nil {
t.Fatalf("CheckHealth over unix socket: %v", err)
}
if health.Status != "ok" {
t.Fatalf("health = %+v", health)
}
// A real workflow, started and read back over the socket.
run, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
ID: "wf-over-socket",
TaskQueue: "default",
}, "Demo")
if err != nil {
t.Fatalf("ExecuteWorkflow over unix socket: %v", err)
}
if run.GetID() != "wf-over-socket" {
t.Fatalf("run id = %q", run.GetID())
}
desc, err := c.DescribeWorkflow(ctx, "wf-over-socket", run.GetRunID())
if err != nil {
t.Fatalf("DescribeWorkflow over unix socket: %v", err)
}
if desc.WorkflowID != "wf-over-socket" {
t.Fatalf("described %+v", desc)
}
}
// The socket is unlinked when the engine stops, so a restart at the same
// path binds cleanly.
func TestE2E_UnixSocketRebind(t *testing.T) {
ctx := context.Background()
dir := t.TempDir()
sock := filepath.Join(dir, "tasks.sock")
for i := 0; i < 3; i++ {
emb, err := tasks.Embed(ctx, tasks.EmbedConfig{Address: sock, DataDir: dir})
if err != nil {
t.Fatalf("bind %d on %s: %v", i, sock, err)
}
if err := emb.Stop(ctx); err != nil {
t.Fatalf("stop %d: %v", i, err)
}
}
}
pkg/tasks/taskqueues_test.go
+2 -2
View File
@@ -13,7 +13,7 @@ import (
)
func TestTaskQueuesAndWorkers(t *testing.T) {
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0})
emb, err := Embed(context.Background(), EmbedConfig{Address: ""})
if err != nil {
t.Fatalf("embed: %v", err)
}
@@ -123,7 +123,7 @@ func TestTaskQueuesAndWorkers(t *testing.T) {
}
func TestWorkflowHistoryAndQuery(t *testing.T) {
emb, err := Embed(context.Background(), EmbedConfig{ZAPPort: 0})
emb, err := Embed(context.Background(), EmbedConfig{Address: ""})
if err != nil {
t.Fatalf("embed: %v", err)
}
pkg/tasks/types.go
+7 -7
View File
@@ -23,12 +23,12 @@ type NamespaceInfo struct {
}
type NamespaceCfg struct {
WorkflowExecutionRetentionTtl string `json:"workflowExecutionRetentionTtl"` // "720h"
APSLimit int `json:"apsLimit"` // actions per second
HistoryArchivalState string `json:"historyArchivalState,omitempty"`
HistoryArchivalUri string `json:"historyArchivalUri,omitempty"`
VisibilityArchivalState string `json:"visibilityArchivalState,omitempty"`
VisibilityArchivalUri string `json:"visibilityArchivalUri,omitempty"`
WorkflowExecutionRetentionTtl string `json:"workflowExecutionRetentionTtl"` // "720h"
APSLimit int `json:"apsLimit"` // actions per second
HistoryArchivalState string `json:"historyArchivalState,omitempty"`
HistoryArchivalUri string `json:"historyArchivalUri,omitempty"`
VisibilityArchivalState string `json:"visibilityArchivalState,omitempty"`
VisibilityArchivalUri string `json:"visibilityArchivalUri,omitempty"`
CustomData map[string]string `json:"customData,omitempty"`
}
@@ -94,7 +94,7 @@ type ScheduleAction struct {
}
type ScheduleState struct {
Paused bool `json:"paused"`
Paused bool `json:"paused"`
Note string `json:"note,omitempty"`
}