diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6cf67bd..9a02bc7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,6 +22,7 @@ jobs: - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: go-version: ${{ env.GO_VERSION }} + check-latest: true cache: true - name: Formatting run: test -z "$(gofmt -l .)" @@ -61,6 +62,7 @@ jobs: - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: go-version: ${{ env.GO_VERSION }} + check-latest: true cache: true - name: Ordinary suite run: go test -count=1 -p 1 -parallel 4 ./... @@ -88,6 +90,7 @@ jobs: - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: go-version: ${{ env.GO_VERSION }} + check-latest: true cache: true - name: Race suite run: make test @@ -100,6 +103,7 @@ jobs: - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 with: go-version: ${{ env.GO_VERSION }} + check-latest: true cache: true - name: govulncheck run: go run golang.org/x/vuln/cmd/govulncheck@latest ./... diff --git a/README.md b/README.md index 73f5d3a..7ecd7ad 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,52 @@ in one deterministic event-key helper used by `WaitFor`, `Deliver`, and `GetCurrentRun`, then `Deliver` to the returned ID. The run can settle between those operations, so `ErrTerminal` is an expected race to handle explicitly. +Application code may also wait for the next durable event without creating a +command. Establish the watch before reading the application projection; an +event is then either included in the baseline and visible to that read, or is +returned by `Next`: + +```go +watch, err := receiptChanged.Watch(ctx, runtime, runID) +if errors.Is(err, flow.ErrTerminal) || errors.Is(err, flow.ErrNotFound) { + return readReceipt(ctx) +} +if err != nil { + return err +} +defer watch.Close() + +receipt, err := readReceipt(ctx) +if err != nil || receipt.Ready { + return err +} +for { + _, _, err := watch.Next(ctx) + if errors.Is(err, flow.ErrTerminal) { + return readReceipt(ctx) + } + if err != nil { + return err + } + receipt, err = readReceipt(ctx) + if err != nil || receipt.Ready { + return err + } +} +``` + +`Event.Watch` is a broadcast inspection API, not a subscription worker: it +creates no command, lease, acknowledgement, callback, or connection per +waiter. `Next` reads the journal and then waits without polling. PostgreSQL +notifications carry only the run ID and wake that durable read; listener +startup/reconnect performs catch-up. A watch may be created before +`Runtime.Run` to close a startup race, but `Next` cannot receive a future-event +wake until the listener starts, so always use a bounded context. Every runtime +that writes a watched run must keep notifications enabled, and application +tables remain the response authority. `History` reads retained facts, +`AwaitRun` waits for terminal run state, and `GetResult` reads one successful +command result; none of them consumes an event. + Positive durable durations may be fractional: Flow rounds them upward once to the next whole millisecond before fingerprinting or persistence. Zero and negative values retain each option's validation rules. diff --git a/compile_contract_test.go b/compile_contract_test.go index 87312f3..c42d9c6 100644 --- a/compile_contract_test.go +++ b/compile_contract_test.go @@ -1,6 +1,7 @@ package flow import ( + "context" "go/ast" "go/parser" "go/token" @@ -12,6 +13,17 @@ import ( "testing" ) +func TestEventWatchCompileContract(t *testing.T) { + event := DefineEvent[string]("compile.watch") + var watchFn func(context.Context, *Runtime, RunID) (*EventWatch[string], error) = event.Watch + var watch *EventWatch[string] + var nextFn func(context.Context) (string, string, error) = watch.Next + var closeFn func() = watch.Close + _ = watchFn + _ = nextFn + _ = closeFn +} + func TestRemovedPublicAPINamesStayRemoved(t *testing.T) { _, currentFile, _, ok := runtime.Caller(0) if !ok { diff --git a/event_gate_test.go b/event_gate_test.go index ca6fac8..78156a4 100644 --- a/event_gate_test.go +++ b/event_gate_test.go @@ -543,7 +543,10 @@ func TestEventGatedCommandsRemainLiveUntilTerminal(t *testing.T) { t.Fatalf("wait expiry trace=%+v", trace.Commands) } - deadline, err := parent.Enqueue(ctx, runtime, "lifecycle/deadline", false, WithRunDeadline(250*time.Millisecond)) + // Leave enough headroom for the race-instrumented parent settlement to + // durably create the gated child before this case exercises run-deadline + // cancellation of that child. + deadline, err := parent.Enqueue(ctx, runtime, "lifecycle/deadline", false, WithRunDeadline(time.Second)) if err != nil { t.Fatal(err) } diff --git a/event_wake.go b/event_wake.go new file mode 100644 index 0000000..0ad8ca6 --- /dev/null +++ b/event_wake.go @@ -0,0 +1,110 @@ +package flow + +import ( + "context" + "sync" + + "github.com/goware/flow/internal/uuid" +) + +// eventWakeHub coalesces run-scoped notification hints. It owns no goroutines: +// EventWatch.Next blocks the caller's goroutine on the returned channel. +type eventWakeHub struct { + mu sync.Mutex + entries map[uuid.UUID]*eventWakeEntry + closed bool + ctx context.Context + cancel context.CancelFunc +} + +type eventWakeEntry struct { + ready chan struct{} + watchers int +} + +func newEventWakeHub() *eventWakeHub { + ctx, cancel := context.WithCancel(context.Background()) + return &eventWakeHub{entries: make(map[uuid.UUID]*eventWakeEntry), ctx: ctx, cancel: cancel} +} + +func (hub *eventWakeHub) register(runID uuid.UUID) (context.Context, bool) { + hub.mu.Lock() + defer hub.mu.Unlock() + if hub.closed { + return nil, false + } + entry := hub.entries[runID] + if entry == nil { + entry = &eventWakeEntry{ready: make(chan struct{})} + hub.entries[runID] = entry + } + entry.watchers++ + return hub.ctx, true +} + +func (hub *eventWakeHub) unregister(runID uuid.UUID) { + hub.mu.Lock() + defer hub.mu.Unlock() + entry := hub.entries[runID] + if entry == nil { + return + } + entry.watchers-- + if entry.watchers == 0 { + delete(hub.entries, runID) + } +} + +func (hub *eventWakeHub) snapshot(runID uuid.UUID) (<-chan struct{}, bool) { + hub.mu.Lock() + defer hub.mu.Unlock() + if hub.closed { + return nil, false + } + entry := hub.entries[runID] + if entry == nil { + return nil, false + } + return entry.ready, true +} + +func (hub *eventWakeHub) signal(runID uuid.UUID) { + hub.mu.Lock() + defer hub.mu.Unlock() + if hub.closed { + return + } + entry := hub.entries[runID] + if entry == nil { + return + } + close(entry.ready) + entry.ready = make(chan struct{}) +} + +func (hub *eventWakeHub) signalAll() { + hub.mu.Lock() + defer hub.mu.Unlock() + if hub.closed { + return + } + for _, entry := range hub.entries { + close(entry.ready) + entry.ready = make(chan struct{}) + } +} + +func (hub *eventWakeHub) close() { + hub.mu.Lock() + if hub.closed { + hub.mu.Unlock() + return + } + hub.closed = true + for _, entry := range hub.entries { + close(entry.ready) + } + clear(hub.entries) + hub.mu.Unlock() + hub.cancel() +} diff --git a/event_watch.go b/event_watch.go new file mode 100644 index 0000000..c980d15 --- /dev/null +++ b/event_watch.go @@ -0,0 +1,189 @@ +package flow + +import ( + "context" + "sync" + "sync/atomic" + + "github.com/goware/flow/internal/store/journalcodec" + "github.com/goware/flow/internal/uuid" +) + +// EventWatch observes future durable application events of one definition in +// one run. It is a broadcast reader: it does not consume or acknowledge an +// event and holds no database connection while waiting. +// +// Call Next sequentially. Concurrent Next calls return ErrInvalidState. +// Always call Close; it is safe to call more than once. An EventWatch must not +// be copied. +type EventWatch[T any] struct { + runtime *Runtime + event Event[T] + runID uuid.UUID + cursor int64 + + closeOnce sync.Once + closed atomic.Bool + inNext atomic.Bool + lifetime context.Context + cancel context.CancelFunc +} + +// Watch starts after the run's current journal head. It requires notifications +// on every runtime that may write the watched run. Establish the watch before +// reading the application's projection to close the application read race. +// A watch may be created before runtime.Run starts; until the listener starts +// and performs its catch-up wake, callers must bound Next with a context. +func (event Event[T]) Watch(ctx context.Context, runtime *Runtime, id RunID) (*EventWatch[T], error) { + if event.err != nil || event.def == nil || event.def.Namespace != "application" { + return nil, newError(ErrInvalid, "watch", "event", eventName(event.def), "invalid event definition") + } + if runtime == nil { + return nil, newError(ErrInvalid, "watch", "runtime", "", "runtime is nil") + } + runID, err := parseRunID(id) + if err != nil { + return nil, err + } + runtime.mu.RLock() + closed := runtime.closed || runtime.lifecycle == runtimeStopping || runtime.lifecycle == runtimeStopped + notifications := runtime.notifications + runtime.mu.RUnlock() + if closed { + return nil, newError(ErrClosed, "watch", "runtime", "", "runtime is closed") + } + if !notifications { + return nil, newError(ErrInvalid, "watch", "runtime", "", "event watches require notifications") + } + runtimeLifetime, registered := runtime.eventWakes.register(runID) + if !registered { + return nil, newError(ErrClosed, "watch", "runtime", "", "runtime is closed") + } + lifetime, cancel := context.WithCancel(runtimeLifetime) + watch := &EventWatch[T]{runtime: runtime, event: event, runID: runID, lifetime: lifetime, cancel: cancel} + queryCtx, cancelQuery := context.WithCancel(ctx) + stopLifetimeCancel := context.AfterFunc(lifetime, cancelQuery) + cursor, err := runtime.store.OpenEventWatch(queryCtx, runID) + stopLifetimeCancel() + cancelQuery() + if err != nil { + if lifetime.Err() != nil { + watch.Close() + return nil, newError(ErrClosed, "watch", "runtime", "", "runtime is closed") + } + watch.Close() + return nil, err + } + if isTerminalStoreRunStatus(cursor.Status) { + watch.Close() + return nil, newError(ErrTerminal, "watch", "run", string(id), "run is terminal") + } + if !isActiveStoreRunStatus(cursor.Status) { + watch.Close() + return nil, newError(ErrInvalidState, "watch", "run", string(id), "stored run status is unknown") + } + if _, active := runtime.eventWakes.snapshot(runID); !active { + watch.Close() + return nil, newError(ErrClosed, "watch", "runtime", "", "runtime is closed") + } + watch.cursor = cursor.Position + return watch, nil +} + +// Next returns the next matching durable application event after the watch +// baseline. It waits only for notification/reconnect hints, Close, runtime +// shutdown, or ctx; it performs no periodic polling. +func (watch *EventWatch[T]) Next(ctx context.Context) (string, T, error) { + var zero T + if watch == nil || watch.runtime == nil { + return "", zero, newError(ErrInvalid, "next", "event watch", "", "watch is nil") + } + if watch.closed.Load() { + return "", zero, newError(ErrClosed, "next", "event watch", "", "watch is closed") + } + if !watch.inNext.CompareAndSwap(false, true) { + return "", zero, newError(ErrInvalidState, "next", "event watch", "", "concurrent Next calls are invalid") + } + defer watch.inNext.Store(false) + + for { + ready, active := watch.runtime.eventWakes.snapshot(watch.runID) + if !active || watch.closed.Load() { + return "", zero, newError(ErrClosed, "next", "event watch", "", "watch is closed") + } + queryCtx, cancelQuery := context.WithCancel(ctx) + stopLifetimeCancel := context.AfterFunc(watch.lifetime, cancelQuery) + result, err := watch.runtime.store.ReadEventWatch(queryCtx, watch.runID, watch.cursor, watch.event.def.Name) + stopLifetimeCancel() + cancelQuery() + if err != nil { + if watch.lifetime.Err() != nil { + return "", zero, newError(ErrClosed, "next", "event watch", "", "watch is closed") + } + return "", zero, err + } + if watch.closed.Load() { + return "", zero, newError(ErrClosed, "next", "event watch", "", "watch is closed") + } + if _, active := watch.runtime.eventWakes.snapshot(watch.runID); !active { + return "", zero, newError(ErrClosed, "next", "event watch", "", "runtime is closed") + } + if !result.RunFound { + return "", zero, newError(ErrTerminal, "next", "run", watch.runID.String(), "watched run no longer exists") + } + if result.Found { + decoded, err := journalcodec.DecodeApplicationEvent(result.Body) + if err != nil { + return "", zero, newError(ErrInvalidState, "decode", "event", watch.event.def.Name, "stored event body is invalid") + } + value, err := watch.event.def.Payload.Decode(decoded.Payload) + if err != nil { + return "", zero, newError(ErrInvalidState, "decode", "event payload", watch.event.def.Name, "stored payload does not match its definition") + } + payload, ok := value.(T) + if !ok { + return "", zero, newError(ErrInvalidState, "decode", "event payload", watch.event.def.Name, "stored payload has an incompatible type") + } + watch.cursor = result.Position + return result.Key, payload, nil + } + if isTerminalStoreRunStatus(result.Status) { + return "", zero, newError(ErrTerminal, "next", "run", watch.runID.String(), "run is terminal") + } + if !isActiveStoreRunStatus(result.Status) { + return "", zero, newError(ErrInvalidState, "next", "run", watch.runID.String(), "stored run status is unknown") + } + select { + case <-ctx.Done(): + return "", zero, ctx.Err() + case <-watch.lifetime.Done(): + return "", zero, newError(ErrClosed, "next", "event watch", "", "watch is closed") + case <-ready: + } + } +} + +// Close removes the local registration and unblocks a waiting Next. +func (watch *EventWatch[T]) Close() { + if watch == nil { + return + } + watch.closeOnce.Do(func() { + watch.closed.Store(true) + if watch.cancel != nil { + watch.cancel() + } + if watch.runtime != nil { + watch.runtime.eventWakes.unregister(watch.runID) + } + }) +} + +func isActiveStoreRunStatus(status string) bool { + return status == string(RunStatusRunning) || status == string(RunStatusFailing) +} + +func isTerminalStoreRunStatus(status string) bool { + return status == string(RunStatusSucceeded) || status == string(RunStatusFailed) || + status == string(RunStatusCancelled) || status == string(RunStatusExpired) +} diff --git a/event_watch_test.go b/event_watch_test.go new file mode 100644 index 0000000..69ec171 --- /dev/null +++ b/event_watch_test.go @@ -0,0 +1,1186 @@ +package flow + +import ( + "context" + "crypto/sha256" + "errors" + goruntime "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/goware/flow/internal/fault" + "github.com/goware/flow/internal/pgschema" + "github.com/goware/flow/internal/store" + "github.com/goware/flow/internal/testpg" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type watchedPayload struct { + Value string `json:"value"` +} + +type eventWatchReadContextKey struct{} + +type eventWatchReadTracer struct { + mu sync.Mutex + once sync.Once + done chan struct{} + reads int +} + +func (tracer *eventWatchReadTracer) TraceQueryStart( + ctx context.Context, + _ *pgx.Conn, + data pgx.TraceQueryStartData, +) context.Context { + if strings.Contains(data.SQL, "LEFT JOIN LATERAL") && strings.Contains(data.SQL, "position>$2") { + return context.WithValue(ctx, eventWatchReadContextKey{}, true) + } + return ctx +} + +func (tracer *eventWatchReadTracer) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryEndData) { + if data.Err == nil && ctx.Value(eventWatchReadContextKey{}) == true { + tracer.mu.Lock() + tracer.reads++ + tracer.mu.Unlock() + tracer.once.Do(func() { close(tracer.done) }) + } +} + +func (tracer *eventWatchReadTracer) readCount() int { + tracer.mu.Lock() + defer tracer.mu.Unlock() + return tracer.reads +} + +func TestEventWatchCrossRuntimeOrderAndTerminal(t *testing.T) { + t.Parallel() + database := testpg.Open(t) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[watchedPayload]("watch.changed") + otherEvent := DefineEvent[None]("watch.other") + root := DefineCommand[None, None]("watch.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema), WithNotifications(true)) + if err != nil { + t.Fatal(err) + } + observer := &recordingObserver{} + reader, err := New(database.DB, WithSchema(database.Schema), WithNotifications(true), + WithObserver(observer), WithPollInterval(5*time.Second)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, writer, "watch/order", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + if err := event.Deliver(ctx, writer, run.RunID, "historical", watchedPayload{Value: "old"}); err != nil { + t.Fatal(err) + } + + watch, err := event.Watch(ctx, reader, run.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + cancel, runResult := startRuntime(t, reader) + defer stopRuntime(t, cancel, runResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + + if err := otherEvent.Deliver(ctx, writer, run.RunID, "ignored", None{}); err != nil { + t.Fatal(err) + } + if err := event.Deliver(ctx, writer, run.RunID, "second", watchedPayload{Value: "two"}); err != nil { + t.Fatal(err) + } + if err := event.Deliver(ctx, writer, run.RunID, "third", watchedPayload{Value: "three"}); err != nil { + t.Fatal(err) + } + for _, want := range []struct { + key string + value string + }{{"second", "two"}, {"third", "three"}} { + nextCtx, cancelNext := context.WithTimeout(ctx, 2*time.Second) + key, payload, nextErr := watch.Next(nextCtx) + cancelNext() + if nextErr != nil || key != want.key || payload.Value != want.value { + t.Fatalf("Next() = %q, %#v, %v; want %q/%q", key, payload, nextErr, want.key, want.value) + } + } + + tx, err := database.DB.Conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + txClient := writer.InTx(tx) + if err := event.Deliver(ctx, txClient, run.RunID, "final", watchedPayload{Value: "last"}); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + if err := CancelRun(ctx, txClient, run.RunID, "watch complete"); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + if err := tx.Commit(ctx); err != nil { + t.Fatal(err) + } + finalCtx, cancelFinal := context.WithTimeout(ctx, 2*time.Second) + key, payload, err := watch.Next(finalCtx) + cancelFinal() + if err != nil || key != "final" || payload.Value != "last" { + t.Fatalf("Next(event before terminal) = %q, %#v, %v", key, payload, err) + } + terminalCtx, cancelTerminal := context.WithTimeout(ctx, 2*time.Second) + _, _, err = watch.Next(terminalCtx) + cancelTerminal() + if !errors.Is(err, ErrTerminal) { + t.Fatalf("Next(terminal) error = %v", err) + } +} + +func TestEventWatchWorkerStagedEmitAndRejectedCommit(t *testing.T) { + t.Parallel() + database := testpg.Open(t) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[watchedPayload]("watch.staged") + success := DefineCommand[None, None]("watch.staged.success", 1, WithRetry(Attempts(1))) + rejected := DefineCommand[None, None]("watch.staged.rejected", 1, WithRetry(Attempts(1))) + worker, err := New(database.DB, WithSchema(database.Schema), WithWorkerConcurrency(2), WithPollInterval(5*time.Second)) + if err != nil { + t.Fatal(err) + } + if err := worker.Register( + Handle(success, func(_ context.Context, work *Work[None]) (None, error) { + if err := Emit(work, event, "success", watchedPayload{Value: "committed"}); err != nil { + return None{}, err + } + return None{}, nil + }), + Handle(rejected, func(_ context.Context, work *Work[None]) (None, error) { + if err := Emit(work, event, "rejected", watchedPayload{Value: "rolled-back"}); err != nil { + return None{}, err + } + return None{}, nil + }, WithCommit(func(context.Context, Tx, Commit[None, None]) error { + return NoRetry(errors.New("reject application commit")) + })), + ); err != nil { + t.Fatal(err) + } + observer := &recordingObserver{} + reader, err := New(database.DB, WithSchema(database.Schema), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + cancelReader, readerResult := startRuntime(t, reader) + defer stopRuntime(t, cancelReader, readerResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + + successRun, err := success.Enqueue(ctx, worker, "watch/staged/success", None{}) + if err != nil { + t.Fatal(err) + } + rejectedRun, err := rejected.Enqueue(ctx, worker, "watch/staged/rejected", None{}) + if err != nil { + t.Fatal(err) + } + successWatch, err := event.Watch(ctx, reader, successRun.RunID) + if err != nil { + t.Fatal(err) + } + defer successWatch.Close() + rejectedWatch, err := event.Watch(ctx, reader, rejectedRun.RunID) + if err != nil { + t.Fatal(err) + } + defer rejectedWatch.Close() + cancelWorker, workerResult := startRuntime(t, worker) + defer stopRuntime(t, cancelWorker, workerResult) + + nextCtx, cancelNext := context.WithTimeout(ctx, 3*time.Second) + key, payload, err := successWatch.Next(nextCtx) + cancelNext() + if err != nil || key != "success" || payload.Value != "committed" { + t.Fatalf("successful staged Next() = %q, %#v, %v", key, payload, err) + } + rejectedCtx, cancelRejected := context.WithTimeout(ctx, 3*time.Second) + _, _, err = rejectedWatch.Next(rejectedCtx) + cancelRejected() + if !errors.Is(err, ErrTerminal) { + t.Fatalf("rejected staged Next error = %v", err) + } + var rejectedEvents int + if err := database.DB.Conn.QueryRow(ctx, `SELECT count(*) FROM `+pgschema.Table(database.Schema, "flow_journal")+` + WHERE run_id=$1 AND event_class='application'`, rejectedRun.RunID).Scan(&rejectedEvents); err != nil { + t.Fatal(err) + } + if rejectedEvents != 0 { + t.Fatalf("rejected WithCommit retained %d application events", rejectedEvents) + } +} + +func TestEventWatchHasNoPeriodicOrUnrelatedRunReads(t *testing.T) { + t.Parallel() + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.idle") + root := DefineCommand[None, None]("watch.idle.root", 1) + observer := &recordingObserver{} + runtime, err := New(database.DB, WithSchema(database.Schema), WithPollInterval(time.Millisecond), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, runtime, "watch/idle", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + other, err := root.Enqueue(ctx, runtime, "watch/idle/other", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + watch, err := event.Watch(ctx, runtime, run.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + cancelRun, runResult := startRuntime(t, runtime) + defer stopRuntime(t, cancelRun, runResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + waitCtx, cancelWait := context.WithCancel(ctx) + result := make(chan error, 1) + go func() { + _, _, nextErr := watch.Next(waitCtx) + result <- nextErr + }() + select { + case <-tracer.done: + case <-time.After(time.Second): + t.Fatal("Next did not perform its initial read") + } + initial := tracer.readCount() + time.Sleep(75 * time.Millisecond) + if got := tracer.readCount(); got != initial { + t.Fatalf("idle Next reads = %d, want %d", got, initial) + } + if err := event.Deliver(ctx, runtime, other.RunID, "other", None{}); err != nil { + t.Fatal(err) + } + time.Sleep(50 * time.Millisecond) + if got := tracer.readCount(); got != initial { + t.Fatalf("unrelated run caused %d reads, want %d", got, initial) + } + if err := event.Deliver(ctx, runtime, run.RunID, "target", None{}); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(time.Second) + for tracer.readCount() == initial && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := tracer.readCount(); got != initial+1 { + t.Fatalf("targeted signal reads = %d, want %d", got, initial+1) + } + if err := <-result; err != nil { + t.Fatalf("same-runtime Next error = %v", err) + } + cancelWait() +} + +func TestEventWatchEventOnlyCommitEmitsOneNotificationStatement(t *testing.T) { + recorder := &queryRecorder{} + database := testpg.OpenWithQueryTracer(t, recorder) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.protocol") + root := DefineCommand[None, None]("watch.protocol.root", 1) + runtime, err := New(database.DB, WithSchema(database.Schema), WithNotifications(true)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, runtime, "watch/protocol", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + recorder.reset() + if err := event.Deliver(ctx, runtime, run.RunID, "event", None{}); err != nil { + t.Fatal(err) + } + queries := recorder.snapshot() + notifications := 0 + for _, query := range queries { + if strings.Contains(query, "pg_notify") { + notifications++ + } + } + if notifications != 1 { + t.Fatalf("pg_notify statements = %d, want 1: %#v", notifications, queries) + } + listener := openNotificationListener(t, database, runtime) + gated, err := root.Enqueue(ctx, runtime, "watch/protocol/gated", None{}, WaitFor(event, "release"), Within(time.Hour)) + if err != nil { + t.Fatal(err) + } + recorder.reset() + if err := event.Deliver(ctx, runtime, gated.RunID, "release", None{}); err != nil { + t.Fatal(err) + } + readyQueries := recorder.snapshot() + readyNotifications := 0 + for _, query := range readyQueries { + if strings.Contains(query, "pg_notify") { + readyNotifications++ + } + } + if readyNotifications != 1 { + t.Fatalf("ready-event pg_notify statements = %d, want 1: %#v", readyNotifications, readyQueries) + } + waitForNotificationHintKind(t, listener, gated.RunID, store.NotificationRun, 2*time.Second) + assertNoNotification(t, listener, 150*time.Millisecond) + foldedRun, err := root.Enqueue(ctx, runtime, "watch/protocol/folded", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + tx, err := database.DB.Conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + txClient := runtime.InTx(tx) + if err := event.Deliver(ctx, txClient, foldedRun.RunID, "event", None{}); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + if err := CancelRun(ctx, txClient, foldedRun.RunID, "fold notification hints"); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + if err := tx.Commit(ctx); err != nil { + t.Fatal(err) + } + waitForNotificationHintKind(t, listener, foldedRun.RunID, store.NotificationEvent, 2*time.Second) + assertNoNotification(t, listener, 150*time.Millisecond) + + withoutHints, err := New(database.DB, WithSchema(database.Schema), WithNotifications(false)) + if err != nil { + t.Fatal(err) + } + withoutHintsRun, err := root.Enqueue(ctx, withoutHints, "watch/protocol/without-hints", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + recorder.reset() + if err := event.Deliver(ctx, withoutHints, withoutHintsRun.RunID, "event", None{}); err != nil { + t.Fatal(err) + } + withoutHintQueries := recorder.snapshot() + if len(queries) != len(withoutHintQueries)+1 { + t.Fatalf("event hint statement delta = %d enabled/%d disabled; want exactly one", len(queries), len(withoutHintQueries)) + } + for _, query := range withoutHintQueries { + if strings.Contains(query, "pg_notify") { + t.Fatalf("notification-disabled delivery issued pg_notify: %#v", withoutHintQueries) + } + } + t.Logf("event-only delivery query statements=%d enabled/%d disabled", len(queries), len(withoutHintQueries)) +} + +func TestEventWatchDisabledWriterDoesNotPromiseWake(t *testing.T) { + t.Parallel() + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.disabled_writer") + root := DefineCommand[None, None]("watch.disabled_writer.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema), WithNotifications(false)) + if err != nil { + t.Fatal(err) + } + observer := &recordingObserver{} + reader, err := New(database.DB, WithSchema(database.Schema), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, writer, "watch/disabled-writer", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + watch, err := event.Watch(ctx, reader, run.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + cancelRun, runResult := startRuntime(t, reader) + defer stopRuntime(t, cancelRun, runResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + + nextCtx, cancelNext := context.WithTimeout(ctx, 150*time.Millisecond) + result := make(chan error, 1) + go func() { + _, _, nextErr := watch.Next(nextCtx) + result <- nextErr + }() + select { + case <-tracer.done: + case <-time.After(time.Second): + t.Fatal("Next did not perform its initial read") + } + if err := event.Deliver(ctx, writer, run.RunID, "persisted-without-hint", None{}); err != nil { + t.Fatal(err) + } + if err := <-result; !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("disabled-writer Next error = %v", err) + } + cancelNext() + runID, err := parseRunID(run.RunID) + if err != nil { + t.Fatal(err) + } + reader.eventWakes.signal(runID) + retryCtx, cancelRetry := context.WithTimeout(ctx, time.Second) + key, _, err := watch.Next(retryCtx) + cancelRetry() + if err != nil || key != "persisted-without-hint" { + t.Fatalf("Next(after explicit catch-up) = %q, %v", key, err) + } +} + +func TestEventWatchRejectsCorruptBodyAndTreatsPruningAsTerminal(t *testing.T) { + t.Parallel() + database := testpg.Open(t) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.corruption") + root := DefineCommand[None, None]("watch.corruption.root", 1) + runtime, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + corruptRun, err := root.Enqueue(ctx, runtime, "watch/corrupt", None{}, WithLiveKey(), WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + corruptWatch, err := event.Watch(ctx, runtime, corruptRun.RunID) + if err != nil { + t.Fatal(err) + } + defer corruptWatch.Close() + if err := event.Deliver(ctx, runtime, corruptRun.RunID, "bad", None{}); err != nil { + t.Fatal(err) + } + badBody := []byte(`{"payload":{},"v":99}`) + digest := sha256.Sum256(badBody) + if _, err := database.DB.Conn.Exec(ctx, `UPDATE `+pgschema.Table(database.Schema, "flow_journal")+` + SET body=$3,body_hash=$4 WHERE run_id=$1 AND event_key=$2 AND event_class='application'`, + corruptRun.RunID, "bad", badBody, digest[:]); err != nil { + t.Fatal(err) + } + if _, _, err := corruptWatch.Next(ctx); !errors.Is(err, ErrInvalidState) { + t.Fatalf("Next(corrupt body) error = %v", err) + } + + prunedRun, err := root.Enqueue(ctx, runtime, "watch/pruned", None{}, WithLiveKey(), WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + prunedWatch, err := event.Watch(ctx, runtime, prunedRun.RunID) + if err != nil { + t.Fatal(err) + } + defer prunedWatch.Close() + if err := CancelRun(ctx, runtime, prunedRun.RunID, "prune watch test"); err != nil { + t.Fatal(err) + } + pruned, err := PruneTerminalRuns(ctx, runtime, time.Now().Add(time.Second), 10) + if err != nil || pruned.Runs < 1 { + t.Fatalf("PruneTerminalRuns() = %#v, %v", pruned, err) + } + if _, _, err := prunedWatch.Next(ctx); !errors.Is(err, ErrTerminal) { + t.Fatalf("Next(pruned run) error = %v", err) + } +} + +func TestEventWatchReconnectCatchUpFindsDisconnectedCommit(t *testing.T) { + t.Parallel() + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.reconnect") + root := DefineCommand[None, None]("watch.reconnect.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + observer := &recordingObserver{} + reader, err := New(database.DB, WithSchema(database.Schema), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + var disconnect atomic.Bool + releaseReconnect := make(chan struct{}) + reader.faults = fault.Func(func(hookCtx context.Context, point fault.Point) error { + if point != fault.NotifyConnect || !disconnect.Load() { + return nil + } + select { + case <-releaseReconnect: + return nil + case <-hookCtx.Done(): + return hookCtx.Err() + } + }) + run, err := root.Enqueue(ctx, writer, "watch/reconnect", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + watch, err := event.Watch(ctx, reader, run.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + cancelRun, runResult := startRuntime(t, reader) + defer stopRuntime(t, cancelRun, runResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + + nextCtx, cancelNext := context.WithTimeout(ctx, 300*time.Millisecond) + defer cancelNext() + next := make(chan struct { + key string + err error + }, 1) + go func() { + key, _, nextErr := watch.Next(nextCtx) + next <- struct { + key string + err error + }{key: key, err: nextErr} + }() + select { + case <-tracer.done: + case <-time.After(time.Second): + t.Fatal("Next did not complete its initial read") + } + disconnect.Store(true) + var terminated bool + if err := database.DB.Conn.QueryRow(ctx, `SELECT pg_terminate_backend(pid) FROM pg_stat_activity + WHERE application_name=$1 AND pid<>pg_backend_pid() LIMIT 1`, + "flow-listener-"+reader.instanceID.String()).Scan(&terminated); err != nil { + t.Fatal(err) + } + if !terminated { + t.Fatal("listener backend was not terminated") + } + waitForObservation(t, observer, "notify_listener", "reconnecting", 1, 2*time.Second) + if err := event.Deliver(ctx, writer, run.RunID, "during-disconnect", None{}); err != nil { + t.Fatal(err) + } + select { + case got := <-next: + if !errors.Is(got.err, context.DeadlineExceeded) { + t.Fatalf("Next(during outage) = %q, %v", got.key, got.err) + } + case <-time.After(time.Second): + t.Fatal("Next did not reach its caller deadline during listener outage") + } + close(releaseReconnect) + waitForObservation(t, observer, "notify_listener", "listening", 2, 2*time.Second) + retryCtx, cancelRetry := context.WithTimeout(ctx, 2*time.Second) + key, _, err := watch.Next(retryCtx) + cancelRetry() + if err != nil || key != "during-disconnect" { + t.Fatalf("Next(after reconnect) = %q, %v", key, err) + } +} + +func TestEventWatchRunReplacementWakesPredecessor(t *testing.T) { + t.Parallel() + database := testpg.Open(t) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.replacement") + root := DefineCommand[None, None]("watch.replacement.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + observer := &recordingObserver{} + reader, err := New(database.DB, WithSchema(database.Schema), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + original, err := root.Enqueue(ctx, writer, "watch/replacement", None{}, WithLiveKey(), WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + watch, err := event.Watch(ctx, reader, original.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + cancelRun, runResult := startRuntime(t, reader) + defer stopRuntime(t, cancelRun, runResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + + next := make(chan error, 1) + go func() { + _, _, nextErr := watch.Next(ctx) + next <- nextErr + }() + replacement, err := root.ReplaceCurrentRun(ctx, writer, original.RunID, "watch/replacement", None{}, + "new generation", WithLiveKey(), WithStartDelay(time.Hour)) + if err != nil || !replacement.Replaced || replacement.RunID == original.RunID { + t.Fatalf("ReplaceCurrentRun() = %#v, %v", replacement, err) + } + select { + case err := <-next: + if !errors.Is(err, ErrTerminal) { + t.Fatalf("predecessor Next error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("replacement did not wake predecessor watch") + } + current, found, err := GetCurrentRun(ctx, writer, root.Name(), "watch/replacement") + if err != nil || !found || current.ID != replacement.RunID { + t.Fatalf("GetCurrentRun() = %#v, %v, %v", current, found, err) + } +} + +func TestEventWatchBroadcastsAcrossRuntimesAndWatchers(t *testing.T) { + t.Parallel() + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.broadcast") + root := DefineCommand[None, None]("watch.broadcast.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, writer, "watch/broadcast", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + type runningRuntime struct { + cancel context.CancelFunc + done <-chan error + } + readers := make([]*Runtime, 2) + running := make([]runningRuntime, len(readers)) + watches := make([]*EventWatch[None], 0, 4) + for index := range readers { + readers[index], err = New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + running[index].cancel, running[index].done = startRuntime(t, readers[index]) + defer stopRuntime(t, running[index].cancel, running[index].done) + for range 2 { + watch, watchErr := event.Watch(ctx, readers[index], run.RunID) + if watchErr != nil { + t.Fatal(watchErr) + } + watches = append(watches, watch) + defer watch.Close() + } + } + initialReads := tracer.readCount() + results := make(chan struct { + key string + err error + }, len(watches)) + for _, watch := range watches { + go func() { + key, _, nextErr := watch.Next(ctx) + results <- struct { + key string + err error + }{key: key, err: nextErr} + }() + } + deadline := time.Now().Add(2 * time.Second) + for tracer.readCount() < initialReads+len(watches) && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := tracer.readCount(); got < initialReads+len(watches) { + t.Fatalf("initial watch reads = %d, want at least %d", got, initialReads+len(watches)) + } + if err := event.Deliver(ctx, writer, run.RunID, "shared", None{}); err != nil { + t.Fatal(err) + } + for range watches { + select { + case got := <-results: + if got.err != nil || got.key != "shared" { + t.Fatalf("broadcast Next() = %q, %v", got.key, got.err) + } + case <-time.After(2 * time.Second): + t.Fatal("broadcast watch did not wake") + } + } +} + +func TestEventWatchMalformedHintPerformsBroadCatchUp(t *testing.T) { + t.Parallel() + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.broad") + root := DefineCommand[None, None]("watch.broad.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema), WithNotifications(false)) + if err != nil { + t.Fatal(err) + } + reader, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, writer, "watch/broad", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + watch, err := event.Watch(ctx, reader, run.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + cancelRun, runResult := startRuntime(t, reader) + defer stopRuntime(t, cancelRun, runResult) + result := make(chan error, 1) + go func() { + key, _, nextErr := watch.Next(ctx) + if nextErr == nil && key != "recovered" { + nextErr = errors.New("unexpected recovered event key") + } + result <- nextErr + }() + select { + case <-tracer.done: + case <-time.After(time.Second): + t.Fatal("Next did not complete its initial read") + } + if err := event.Deliver(ctx, writer, run.RunID, "recovered", None{}); err != nil { + t.Fatal(err) + } + if _, err := database.DB.Conn.Exec(ctx, `SELECT pg_notify($1,$2)`, reader.store.NotificationChannel(), `{"v":99}`); err != nil { + t.Fatal(err) + } + select { + case err := <-result: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("malformed hint did not trigger broad catch-up") + } +} + +func TestEventWatchThousandIdleWatchersDoNotPoll(t *testing.T) { + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.scale") + root := DefineCommand[None, None]("watch.scale.root", 1) + observer := &recordingObserver{} + runtime, err := New(database.DB, WithSchema(database.Schema), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, runtime, "watch/scale", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + readDurableWork := func() (commands, queueRows, leases int) { + t.Helper() + if err := database.DB.Conn.QueryRow(ctx, `SELECT + (SELECT count(*) FROM `+pgschema.Table(database.Schema, "flow_commands")+` WHERE run_id=$1), + (SELECT count(*) FROM `+pgschema.Table(database.Schema, "flow_command_queue")+` WHERE run_id=$1), + (SELECT count(*) FROM `+pgschema.Table(database.Schema, "flow_command_queue")+` + WHERE run_id=$1 AND lease_token IS NOT NULL)`, run.RunID).Scan(&commands, &queueRows, &leases); err != nil { + t.Fatal(err) + } + return commands, queueRows, leases + } + beforeCommands, beforeQueueRows, beforeLeases := readDurableWork() + const count = 1000 + watches := make([]*EventWatch[None], count) + beforeRegistration := goruntime.NumGoroutine() + for index := range watches { + watches[index], err = event.Watch(ctx, runtime, run.RunID) + if err != nil { + t.Fatalf("Watch(%d) error = %v", index, err) + } + } + if added := goruntime.NumGoroutine() - beforeRegistration; added > 4 { + t.Fatalf("watch registration added %d goroutines", added) + } + cancelRun, runResult := startRuntime(t, runtime) + defer stopRuntime(t, cancelRun, runResult) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + listenerDeadline := time.Now().Add(2 * time.Second) + for { + var listeners int + if err := database.DB.Conn.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE application_name=$1`, + "flow-listener-"+runtime.instanceID.String()).Scan(&listeners); err != nil { + t.Fatal(err) + } + if listeners == 1 { + break + } + if time.Now().After(listenerDeadline) { + t.Fatalf("dedicated listener connections = %d, want 1", listeners) + } + time.Sleep(time.Millisecond) + } + + initialReads := tracer.readCount() + waitCtx, cancelWait := context.WithCancel(ctx) + results := make(chan error, count) + for _, watch := range watches { + go func() { + _, _, nextErr := watch.Next(waitCtx) + results <- nextErr + }() + } + deadline := time.Now().Add(5 * time.Second) + for tracer.readCount() < initialReads+count && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := tracer.readCount(); got != initialReads+count { + cancelWait() + t.Fatalf("initial Next reads = %d, want %d", got, initialReads+count) + } + connectionDeadline := time.Now().Add(time.Second) + for database.DB.Conn.Stat().AcquiredConns() != 0 && time.Now().Before(connectionDeadline) { + time.Sleep(time.Millisecond) + } + if acquired := database.DB.Conn.Stat().AcquiredConns(); acquired != 0 { + cancelWait() + t.Fatalf("idle watches retain %d application connections", acquired) + } + readCount := tracer.readCount() + time.Sleep(75 * time.Millisecond) + if got := tracer.readCount(); got != readCount { + cancelWait() + t.Fatalf("idle watch reads grew from %d to %d", readCount, got) + } + afterCommands, afterQueueRows, afterLeases := readDurableWork() + if afterCommands != beforeCommands || afterQueueRows != beforeQueueRows || afterLeases != beforeLeases { + cancelWait() + t.Fatalf("watchers changed durable work: commands %d/%d queue %d/%d leases %d/%d", + beforeCommands, afterCommands, beforeQueueRows, afterQueueRows, beforeLeases, afterLeases) + } + cancelWait() + for range count { + if err := <-results; !errors.Is(err, context.Canceled) { + t.Fatalf("Next cancellation error = %v", err) + } + } + for _, watch := range watches { + watch.Close() + } + runtime.eventWakes.mu.Lock() + entries := len(runtime.eventWakes.entries) + runtime.eventWakes.mu.Unlock() + if entries != 0 { + t.Fatalf("event wake entries after Close = %d", entries) + } +} + +func TestEventWatchValidationCancellationAndClose(t *testing.T) { + t.Parallel() + database := testpg.Open(t) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.validation") + root := DefineCommand[None, None]("watch.validation.root", 1) + runtime, err := New(database.DB, WithSchema(database.Schema), WithNotifications(true)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, runtime, "watch/validation", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + if _, err := event.Watch(ctx, nil, run.RunID); !errors.Is(err, ErrInvalid) { + t.Fatalf("Watch(nil) error = %v", err) + } + var invalidEvent Event[None] + if _, err := invalidEvent.Watch(ctx, runtime, run.RunID); !errors.Is(err, ErrInvalid) { + t.Fatalf("Watch(invalid event) error = %v", err) + } + disabled, err := New(database.DB, WithSchema(database.Schema), WithNotifications(false)) + if err != nil { + t.Fatal(err) + } + if _, err := event.Watch(ctx, disabled, run.RunID); !errors.Is(err, ErrInvalid) { + t.Fatalf("Watch(notifications disabled) error = %v", err) + } + if _, err := event.Watch(ctx, runtime, RunID("bad")); !errors.Is(err, ErrInvalid) { + t.Fatalf("Watch(invalid run ID) error = %v", err) + } + if _, err := event.Watch(ctx, runtime, RunID("00000000-0000-0000-0000-000000000001")); !errors.Is(err, ErrNotFound) { + t.Fatalf("Watch(missing run) error = %v", err) + } + constructorCtx, cancelConstructor := context.WithCancel(ctx) + cancelConstructor() + if _, err := event.Watch(constructorCtx, runtime, run.RunID); !errors.Is(err, context.Canceled) { + t.Fatalf("Watch(cancelled construction) error = %v", err) + } + runtime.eventWakes.mu.Lock() + failedEntries := len(runtime.eventWakes.entries) + runtime.eventWakes.mu.Unlock() + if failedEntries != 0 { + t.Fatalf("failed watch construction retained %d wake entries", failedEntries) + } + + watch, err := event.Watch(ctx, runtime, run.RunID) + if err != nil { + t.Fatal(err) + } + cancelled, cancel := context.WithCancel(ctx) + cancel() + if _, _, err := watch.Next(cancelled); !errors.Is(err, context.Canceled) { + t.Fatalf("Next(cancelled) error = %v", err) + } + if err := event.Deliver(ctx, runtime, run.RunID, "after-cancel", None{}); err != nil { + t.Fatal(err) + } + nextCtx, cancelNext := context.WithTimeout(ctx, time.Second) + key, _, err := watch.Next(nextCtx) + cancelNext() + if err != nil || key != "after-cancel" { + t.Fatalf("reused Next() = %q, %v", key, err) + } + + waiting := make(chan error, 1) + connections := make([]*pgxpool.Conn, 0, runtime.db.Conn.Config().MaxConns) + for range runtime.db.Conn.Config().MaxConns { + connection, acquireErr := runtime.db.Conn.Acquire(ctx) + if acquireErr != nil { + t.Fatal(acquireErr) + } + connections = append(connections, connection) + } + defer func() { + for _, connection := range connections { + connection.Release() + } + }() + go func() { + _, _, waitErr := watch.Next(context.Background()) + waiting <- waitErr + }() + deadline := time.Now().Add(time.Second) + for !watch.inNext.Load() && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if !watch.inNext.Load() { + t.Fatal("Next did not begin") + } + if _, _, err := watch.Next(ctx); !errors.Is(err, ErrInvalidState) { + t.Fatalf("concurrent Next error = %v", err) + } + watch.Close() + watch.Close() + select { + case err := <-waiting: + if !errors.Is(err, ErrClosed) { + t.Fatalf("waiting Next error = %v", err) + } + case <-time.After(time.Second): + t.Fatal("Close did not cancel a Next blocked on database acquisition") + } + for _, connection := range connections { + connection.Release() + } + connections = nil + if _, _, err := watch.Next(ctx); !errors.Is(err, ErrClosed) { + t.Fatalf("Next(closed) error = %v", err) + } + if err := CancelRun(ctx, runtime, run.RunID, "validation complete"); err != nil { + t.Fatal(err) + } + if _, err := event.Watch(ctx, runtime, run.RunID); !errors.Is(err, ErrTerminal) { + t.Fatalf("Watch(terminal run) error = %v", err) + } +} + +func TestEventWatchRuntimeStopCancelsConstruction(t *testing.T) { + t.Parallel() + database := testpg.Open(t) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.construction_stop") + root := DefineCommand[None, None]("watch.construction_stop.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + run, err := root.Enqueue(ctx, writer, "watch/construction-stop", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + reader, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + connections := make([]*pgxpool.Conn, 0, database.DB.Conn.Config().MaxConns) + for range database.DB.Conn.Config().MaxConns { + connection, acquireErr := database.DB.Conn.Acquire(ctx) + if acquireErr != nil { + t.Fatal(acquireErr) + } + connections = append(connections, connection) + } + defer func() { + for _, connection := range connections { + connection.Release() + } + }() + + constructed := make(chan error, 1) + go func() { + _, watchErr := event.Watch(context.Background(), reader, run.RunID) + constructed <- watchErr + }() + deadline := time.Now().Add(time.Second) + for { + reader.eventWakes.mu.Lock() + registered := len(reader.eventWakes.entries) == 1 + reader.eventWakes.mu.Unlock() + if registered { + break + } + if time.Now().After(deadline) { + t.Fatal("Watch did not register before its initial database read") + } + time.Sleep(time.Millisecond) + } + if err := reader.Stop(ctx); err != nil { + t.Fatal(err) + } + select { + case err := <-constructed: + if !errors.Is(err, ErrClosed) { + t.Fatalf("Watch(runtime stopped during construction) error = %v", err) + } + case <-time.After(time.Second): + t.Fatal("runtime stop did not cancel watch construction") + } +} + +func TestEventWatchPreRunCatchUpAndShutdown(t *testing.T) { + t.Parallel() + tracer := &eventWatchReadTracer{done: make(chan struct{})} + database := testpg.OpenWithQueryTracer(t, tracer) + ctx := context.Background() + if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + event := DefineEvent[None]("watch.startup") + root := DefineCommand[None, None]("watch.startup.root", 1) + writer, err := New(database.DB, WithSchema(database.Schema)) + if err != nil { + t.Fatal(err) + } + observer := &recordingObserver{} + reader, err := New(database.DB, WithSchema(database.Schema), WithPollInterval(5*time.Second), WithObserver(observer)) + if err != nil { + t.Fatal(err) + } + var connectAttempts atomic.Int32 + reader.faults = fault.Func(func(_ context.Context, point fault.Point) error { + if point == fault.NotifyConnect && connectAttempts.Add(1) == 1 { + return fault.Injected(point) + } + return nil + }) + run, err := root.Enqueue(ctx, writer, "watch/startup", None{}, WithStartDelay(time.Hour)) + if err != nil { + t.Fatal(err) + } + watch, err := event.Watch(ctx, reader, run.RunID) + if err != nil { + t.Fatal(err) + } + defer watch.Close() + result := make(chan struct { + key string + err error + }, 1) + go func() { + key, _, nextErr := watch.Next(ctx) + result <- struct { + key string + err error + }{key: key, err: nextErr} + }() + select { + case <-tracer.done: + case <-time.After(time.Second): + t.Fatal("Next did not complete its initial durable read") + } + if err := event.Deliver(ctx, writer, run.RunID, "during-startup", None{}); err != nil { + t.Fatal(err) + } + select { + case got := <-result: + t.Fatalf("pre-Run Next returned without a listener: %#v", got) + case <-time.After(75 * time.Millisecond): + } + cancelRun, runResult := startRuntime(t, reader) + waitForObservation(t, observer, "notify_listener", "connect_error", 1, 2*time.Second) + waitForObservation(t, observer, "notify_listener", "listening", 1, 2*time.Second) + select { + case got := <-result: + if got.err != nil || got.key != "during-startup" { + t.Fatalf("startup catch-up Next() = %q, %v", got.key, got.err) + } + case <-time.After(2 * time.Second): + t.Fatal("listener startup did not catch up the watch") + } + stopRuntime(t, cancelRun, runResult) + if _, _, err := watch.Next(ctx); !errors.Is(err, ErrClosed) { + t.Fatalf("Next(stopped runtime) error = %v", err) + } + if _, err := event.Watch(ctx, reader, run.RunID); !errors.Is(err, ErrClosed) { + t.Fatalf("Watch(stopped runtime) error = %v", err) + } +} diff --git a/flow.go b/flow.go index 153a0d7..af557ab 100644 --- a/flow.go +++ b/flow.go @@ -123,6 +123,13 @@ // selected run settles before delivery. Event definitions should name stable // fact kinds; deterministic keys should carry entity and generation identity. // +// [Event.Watch] observes future matching application events without creating +// durable work. Construct the watch before reading the application's own +// projection, then call [EventWatch.Next] sequentially under a bounded context. +// Notification payloads carry only run identity; Next returns data decoded +// from the durable journal. Watches hold no connection and do not poll, so +// every runtime writing a watched run must keep notifications enabled. +// // [Command.ReplaceCurrentRun] atomically cancels an exact expected live-key // generation and creates a distinct successor. Retries can rediscover a // declaration-equivalent successor only after the current run ID differs from diff --git a/hardening_benchmark_test.go b/hardening_benchmark_test.go index 7641824..24b59cb 100644 --- a/hardening_benchmark_test.go +++ b/hardening_benchmark_test.go @@ -444,7 +444,7 @@ func setupExternalEventBenchmark(b *testing.B) (*Runtime, Command[None, None], E if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil { b.Fatal(err) } - runtime, err := New(database.DB, WithSchema(database.Schema), WithNotifications(false), + runtime, err := New(database.DB, WithSchema(database.Schema), WithNotifications(true), WithMaxCommandsPerRun(0)) if err != nil { b.Fatal(err) diff --git a/internal/store/commands.go b/internal/store/commands.go index 5a7a077..02d02a6 100644 --- a/internal/store/commands.go +++ b/internal/store/commands.go @@ -1356,6 +1356,10 @@ func (s *Store) SettleCommandSuccess(ctx context.Context, request CommandSuccess if err := semantic.NotifyRunnableCommands(ctx); err != nil { return SettleResult{}, err } + } else if len(request.Events) > 0 { + if err := semantic.NotifyEventWatchers(ctx); err != nil { + return SettleResult{}, err + } } if err := hook.Hit(ctx, fault.SettleBeforeCommit); err != nil { return SettleResult{}, err diff --git a/internal/store/event_watch.go b/internal/store/event_watch.go new file mode 100644 index 0000000..4f21b50 --- /dev/null +++ b/internal/store/event_watch.go @@ -0,0 +1,105 @@ +package store + +import ( + "bytes" + "context" + "crypto/sha256" + "errors" + "fmt" + + "github.com/goware/flow/internal/flowerr" + "github.com/goware/flow/internal/pgschema" + "github.com/goware/flow/internal/uuid" + "github.com/jackc/pgx/v5" +) + +type EventWatchCursor struct { + Position int64 + Status string +} + +type EventWatchRead struct { + RunFound bool + Status string + Position int64 + Key string + Body []byte + Found bool +} + +// OpenEventWatch captures one run's current journal head. Events through this +// position are historical to the newly constructed watch. +func (s *Store) OpenEventWatch(ctx context.Context, runID uuid.UUID) (EventWatchCursor, error) { + if runID == uuid.Nil { + return EventWatchCursor{}, fmt.Errorf("%w: run ID is nil", flowerr.ErrInvalid) + } + var result EventWatchCursor + err := s.db.Conn.QueryRow(ctx, `SELECT status,next_journal_position-1 + FROM `+pgschema.Table(s.schema, "flow_runs")+` WHERE run_id=$1`, runID). + Scan(&result.Status, &result.Position) + if err != nil { + return EventWatchCursor{}, MapError("open event watch", err) + } + if result.Position < 0 { + return EventWatchCursor{}, fmt.Errorf("%w: event watch cursor is negative", flowerr.ErrInvalidState) + } + return result, nil +} + +// ReadEventWatch returns the first matching application event after cursor and +// the run status in one durable read. A missing run is represented explicitly +// because an already-open watch treats retention pruning as terminal. +func (s *Store) ReadEventWatch( + ctx context.Context, + runID uuid.UUID, + after int64, + eventName string, +) (EventWatchRead, error) { + if runID == uuid.Nil || after < 0 || eventName == "" { + return EventWatchRead{}, fmt.Errorf("%w: event watch read is invalid", flowerr.ErrInvalid) + } + var result EventWatchRead + var position *int64 + var key *string + var body, bodyHash []byte + err := s.db.Conn.QueryRow(ctx, s.readEventWatchSQL(), runID, after, eventName). + Scan(&result.Status, &position, &key, &body, &bodyHash) + if errors.Is(err, pgx.ErrNoRows) { + return EventWatchRead{}, nil + } + if err != nil { + return EventWatchRead{}, MapError("read event watch", err) + } + result.RunFound = true + if position == nil { + if key != nil || body != nil || bodyHash != nil { + return EventWatchRead{}, fmt.Errorf("%w: event watch row is incomplete", flowerr.ErrInvalidState) + } + return result, nil + } + if *position <= after || key == nil || *key == "" || len(bodyHash) != sha256.Size { + return EventWatchRead{}, fmt.Errorf("%w: event watch row is invalid", flowerr.ErrInvalidState) + } + digest := sha256.Sum256(body) + if !bytes.Equal(digest[:], bodyHash) { + return EventWatchRead{}, fmt.Errorf("%w: event watch body hash differs", flowerr.ErrInvalidState) + } + result.Position, result.Key, result.Body, result.Found = *position, *key, append([]byte(nil), body...), true + return result, nil +} + +func (s *Store) readEventWatchSQL() string { + return `SELECT r.status, + next_event.position,next_event.event_key,next_event.body,next_event.body_hash + FROM ` + pgschema.Table(s.schema, "flow_runs") + ` AS r + LEFT JOIN LATERAL ( + SELECT position,event_key,body,body_hash + FROM ` + pgschema.Table(s.schema, "flow_journal") + ` + WHERE run_id=r.run_id AND position>$2 + AND entry_kind='event_recorded' + AND event_namespace='application' AND event_class='application' + AND event_name=$3 + ORDER BY position LIMIT 1 + ) AS next_event ON true + WHERE r.run_id=$1` +} diff --git a/internal/store/event_watch_plan_test.go b/internal/store/event_watch_plan_test.go new file mode 100644 index 0000000..1eb0a2e --- /dev/null +++ b/internal/store/event_watch_plan_test.go @@ -0,0 +1,79 @@ +package store_test + +import ( + "context" + "crypto/sha256" + "fmt" + "strings" + "testing" + "time" + + flow "github.com/goware/flow" + "github.com/goware/flow/internal/pgschema" + "github.com/goware/flow/internal/store" + "github.com/goware/flow/internal/testpg" +) + +func TestEventWatchSparsePostCursorPlan(t *testing.T) { + database := testpg.Open(t) + ctx := context.Background() + if err := flow.Migrate(ctx, database.DB, flow.WithSchema(database.Schema)); err != nil { + t.Fatal(err) + } + runtime, err := flow.New(database.DB, flow.WithSchema(database.Schema), flow.WithNotifications(false)) + if err != nil { + t.Fatal(err) + } + repository, err := store.New(database.DB, database.Schema, false) + if err != nil { + t.Fatal(err) + } + root := flow.DefineCommand[flow.None, flow.None]("watch.plan.root", 1) + body := []byte(`{"payload":{},"v":1}`) + digest := sha256.Sum256(body) + for _, count := range []int{100, 1000, 10000} { + t.Run(fmt.Sprint(count), func(t *testing.T) { + run, enqueueErr := root.Enqueue(ctx, runtime, fmt.Sprintf("watch/plan/%d", count), flow.None{}, flow.WithStartDelay(time.Hour)) + if enqueueErr != nil { + t.Fatal(enqueueErr) + } + if _, insertErr := database.DB.Conn.Exec(ctx, `INSERT INTO `+pgschema.Table(database.Schema, "flow_journal")+` ( + run_id,position,entry_id,entry_kind,recorded_at,event_id,event_namespace,event_name,event_key,event_class,body,body_hash + ) SELECT $1::uuid,series+2,md5(($1::uuid)::text||'/entry/'||series)::uuid,'event_recorded',clock_timestamp(), + md5(($1::uuid)::text||'/event/'||series)::uuid,'application', + CASE WHEN series=$2::integer THEN 'watch.plan.target' ELSE 'watch.plan.noise' END, + 'event/'||series,'application',$3,$4 + FROM generate_series(1,$2::integer) AS series`, run.RunID, count, body, digest[:]); insertErr != nil { + t.Fatal(insertErr) + } + if _, updateErr := database.DB.Conn.Exec(ctx, `UPDATE `+pgschema.Table(database.Schema, "flow_runs")+` + SET next_journal_position=$2::bigint+3 WHERE run_id=$1`, run.RunID, count); updateErr != nil { + t.Fatal(updateErr) + } + rows, explainErr := database.DB.Conn.Query(ctx, `EXPLAIN (ANALYZE,BUFFERS,COSTS OFF) `+ + store.EventWatchReadQueryForTest(repository), run.RunID, int64(2), "watch.plan.target") + if explainErr != nil { + t.Fatal(explainErr) + } + var lines []string + for rows.Next() { + var line string + if err := rows.Scan(&line); err != nil { + rows.Close() + t.Fatal(err) + } + lines = append(lines, line) + } + if err := rows.Err(); err != nil { + rows.Close() + t.Fatal(err) + } + rows.Close() + plan := strings.Join(lines, "\n") + if !strings.Contains(plan, "flow_journal_pkey") || strings.Contains(plan, "Seq Scan on flow_journal") { + t.Fatalf("event-watch plan is not cursor-indexed:\n%s", plan) + } + t.Logf("%d post-cursor rows:\n%s", count, plan) + }) + } +} diff --git a/internal/store/export_test.go b/internal/store/export_test.go index 87ce778..32098d0 100644 --- a/internal/store/export_test.go +++ b/internal/store/export_test.go @@ -24,4 +24,6 @@ func PruneCandidatesQueryForTest(s *Store) string { return s.pruneCandidatesQuer func ProbeCommandsQueryForTest(s *Store) string { return s.probeCommandsSQL() } +func EventWatchReadQueryForTest(s *Store) string { return s.readEventWatchSQL() } + func TraceWaitsQueryForTest(s *Store) string { return s.traceWaitsQuery() } diff --git a/internal/store/ingress.go b/internal/store/ingress.go index 144649b..0b19ad2 100644 --- a/internal/store/ingress.go +++ b/internal/store/ingress.go @@ -1020,6 +1020,8 @@ func (s *Store) EmitLocked(ctx context.Context, semantic *SemanticTx, event Appl if err := semantic.NotifyRunnableCommands(ctx); err != nil { return false, err } + } else if err := semantic.NotifyEventWatchers(ctx); err != nil { + return false, err } return true, nil } diff --git a/internal/store/store.go b/internal/store/store.go index 522f8d8..2a02a63 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -40,7 +40,7 @@ type Store struct { } // New constructs the PostgreSQL store. Notifications controls transactional -// wake hints; correctness never depends on it. +// wake hints. Command scheduling retains polling; EventWatch requires hints. func New(db *pgkit.DB, schema string, notifications bool) (*Store, error) { if db == nil || db.Conn == nil { return nil, fmt.Errorf("%w: database is nil", flowerr.ErrInvalid) @@ -72,20 +72,33 @@ func (s *Store) NotificationChannel() string { return s.notificationChannel } +const ( + NotificationRun = "run" + NotificationEvent = "event" +) + +type NotificationHint struct { + RunID uuid.UUID + Kind string +} + // ParseNotificationHint validates the deliberately tiny, versioned payload. -// A hint is never durable work and is safe to discard; polling remains the -// correctness mechanism for malformed or future versions. -func ParseNotificationHint(payload string) (uuid.UUID, bool) { +// A hint carries identity only; callers must read durable state after it. +func ParseNotificationHint(payload string) (NotificationHint, bool) { var hint struct { V int `json:"v"` Kind string `json:"kind"` Key string `json:"key"` } - if err := json.Unmarshal([]byte(payload), &hint); err != nil || hint.V != 1 || hint.Kind != "run" { - return uuid.Nil, false + if err := json.Unmarshal([]byte(payload), &hint); err != nil || hint.V != 1 || + (hint.Kind != NotificationRun && hint.Kind != NotificationEvent) { + return NotificationHint{}, false } id, err := uuid.Parse(hint.Key) - return id, err == nil + if err != nil { + return NotificationHint{}, false + } + return NotificationHint{RunID: id, Kind: hint.Kind}, true } type SemanticTx struct { @@ -96,7 +109,8 @@ type SemanticTx struct { initialLockedSnapshot *InitialLockedSnapshot closed bool applied bool - notificationSent bool + eventHintSent bool + runHintSent bool notificationOwner *SemanticTx failed bool } @@ -296,6 +310,16 @@ func (tx *SemanticTx) Apply(ctx context.Context, changes PersistedChangeSet) (Ap return ApplyResult{}, fmt.Errorf("%w: journal batch inserted %d of %d rows", flowerr.ErrInvalidState, count, len(copyRows)) } tx.applied = true + // Run terminal entries always wake event watches. Application-event + // operations choose event versus run only after readiness is projected. + for _, entry := range changes.Journal { + if entry.Kind == EventRecorded && entry.EventClass != nil && *entry.EventClass == "run_terminal" { + if err := tx.NotifyEventWatchers(ctx); err != nil { + return ApplyResult{}, err + } + break + } + } return ApplyResult{Journal: cloneJournalRows(rows)}, nil } @@ -314,15 +338,41 @@ func (tx *SemanticTx) NotifyRunnableCommands(ctx context.Context) error { if tx.notificationOwner != nil { notificationState = tx.notificationOwner } - if notificationState.notificationSent { + if notificationState.runHintSent { return nil } - payload := `{"v":1,"kind":"run","key":"` + tx.runID.String() + `"}` + payload := `{"v":1,"kind":"` + NotificationRun + `","key":"` + tx.runID.String() + `"}` if _, err := tx.tx.Exec(ctx, `SELECT pg_notify($1, $2)`, tx.store.notificationChannel, payload); err != nil { tx.failed = true return MapError("notify runnable commands", err) } - notificationState.notificationSent = true + notificationState.runHintSent = true + return nil +} + +// NotifyEventWatchers emits one transactional run-identity hint for durable +// application events and run terminal transitions. A run hint is stronger and +// already wakes event watchers, so it suppresses a later event hint. +func (tx *SemanticTx) NotifyEventWatchers(ctx context.Context) error { + if err := tx.ensureOpen("notify event watchers"); err != nil { + return err + } + if !tx.store.notifications { + return nil + } + notificationState := tx + if tx.notificationOwner != nil { + notificationState = tx.notificationOwner + } + if notificationState.eventHintSent || notificationState.runHintSent { + return nil + } + payload := `{"v":1,"kind":"` + NotificationEvent + `","key":"` + tx.runID.String() + `"}` + if _, err := tx.tx.Exec(ctx, `SELECT pg_notify($1, $2)`, tx.store.notificationChannel, payload); err != nil { + tx.failed = true + return MapError("notify event watchers", err) + } + notificationState.eventHintSent = true return nil } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index f183f6e..e9d5ce2 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -326,8 +326,12 @@ func TestNotificationChannelAndPayload(t *testing.T) { } id := uuid.New() parsed, ok := store.ParseNotificationHint(`{"v":1,"kind":"run","key":"` + id.String() + `"}`) - if !ok || parsed != id { - t.Fatalf("parsed notification=%s/%t want %s", parsed, ok, id) + if !ok || parsed.RunID != id || parsed.Kind != store.NotificationRun { + t.Fatalf("parsed notification=%#v/%t want %s", parsed, ok, id) + } + parsed, ok = store.ParseNotificationHint(`{"v":1,"kind":"event","key":"` + id.String() + `"}`) + if !ok || parsed.RunID != id || parsed.Kind != store.NotificationEvent { + t.Fatalf("parsed event notification=%#v/%t want %s", parsed, ok, id) } for _, invalid := range []string{"", `{}`, `{"v":2,"kind":"run","key":"` + id.String() + `"}`, `{"v":1,"kind":"work","key":"` + id.String() + `"}`} { diff --git a/notification_test.go b/notification_test.go index fac11cb..d321799 100644 --- a/notification_test.go +++ b/notification_test.go @@ -59,9 +59,9 @@ func TestNotificationHintsCommitButDoNotRollback(t *testing.T) { if err != nil { t.Fatalf("WaitForNotification(commit) error = %v", err) } - id, valid := store.ParseNotificationHint(notification.Payload) - if !valid || id.String() != string(exec.RunID) || notification.Channel != channel { - t.Fatalf("notification = %#v, parsed=%s/%t", notification, id, valid) + hint, valid := store.ParseNotificationHint(notification.Payload) + if !valid || hint.RunID.String() != string(exec.RunID) || hint.Kind != store.NotificationRun || notification.Channel != channel { + t.Fatalf("notification = %#v, parsed=%#v/%t", notification, hint, valid) } if len(notification.Payload) > 128 { t.Fatalf("notification payload contains more than a bounded identity hint: %q", notification.Payload) @@ -103,7 +103,7 @@ func TestReplaceCurrentRunNotifiesOnlyForCommittedRunnableSuccessor(t *testing.T if err != nil || !replaced.Replaced { t.Fatalf("replacement = %#v, %v", replaced, err) } - waitForNotificationHint(t, listener, replaced.RunID, 2*time.Second) + waitForNotificationHintKind(t, listener, replaced.RunID, store.NotificationRun, 2*time.Second) rolledBackOriginal, err := command.Enqueue(ctx, runtime, "notify/replace-rollback", None{}, WithLiveKey(), WithStartDelay(time.Hour)) if err != nil { @@ -126,7 +126,7 @@ func TestReplaceCurrentRunNotifiesOnlyForCommittedRunnableSuccessor(t *testing.T assertNoNotification(t, listener, 150*time.Millisecond) } -func TestNotificationHintsOnlyForRunnableTransitions(t *testing.T) { +func TestNotificationHintsForRunnableAndApplicationEventTransitions(t *testing.T) { t.Parallel() database := testpg.Open(t) @@ -162,20 +162,37 @@ func TestNotificationHintsOnlyForRunnableTransitions(t *testing.T) { if err := event.Deliver(ctx, api, exec.RunID, "unrelated", None{}); err != nil { t.Fatalf("Emit(unrelated) error = %v", err) } + waitForNotificationHintKind(t, listener, exec.RunID, store.NotificationEvent, 2*time.Second) + if err := event.Deliver(ctx, api, exec.RunID, "unrelated", None{}); err != nil { + t.Fatalf("Emit(equivalent) error = %v", err) + } + assertNoNotification(t, listener, 150*time.Millisecond) + tx, err := database.DB.Conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + if err := event.Deliver(ctx, api.InTx(tx), exec.RunID, "rolled-back", None{}); err != nil { + _ = tx.Rollback(ctx) + t.Fatal(err) + } + assertNoNotification(t, listener, 100*time.Millisecond) + if err := tx.Rollback(ctx); err != nil { + t.Fatal(err) + } assertNoNotification(t, listener, 150*time.Millisecond) started := time.Now() if err := event.Deliver(ctx, api, exec.RunID, "ready", None{}); err != nil { t.Fatalf("Emit(ready) error = %v", err) } - waitForNotificationHint(t, listener, exec.RunID, 2*time.Second) + waitForNotificationHintKind(t, listener, exec.RunID, store.NotificationRun, 2*time.Second) waitForRunStatus(t, database.Schema, database.DB.Conn, exec.RunID, "succeeded", 2*time.Second) if elapsed := time.Since(started); elapsed >= 2*time.Second { t.Fatalf("event-release notification wake took %s with five-second polling", elapsed) } } -func TestClaimAndTerminalSettlementDoNotNotify(t *testing.T) { +func TestClaimDoesNotNotifyAndTerminalSettlementWakesWatchers(t *testing.T) { t.Parallel() database := testpg.Open(t) @@ -193,7 +210,7 @@ func TestClaimAndTerminalSettlementDoNotNotify(t *testing.T) { if err != nil { t.Fatalf("Enqueue() error = %v", err) } - waitForNotificationHint(t, listener, exec.RunID, 2*time.Second) + waitForNotificationHintKind(t, listener, exec.RunID, store.NotificationRun, 2*time.Second) candidates, err := runtime.store.ProbeCommands(ctx, []store.CommandKind{{Name: command.Name(), Version: command.Version()}}, 1) if err != nil || len(candidates) != 1 { @@ -211,7 +228,7 @@ func TestClaimAndTerminalSettlementDoNotNotify(t *testing.T) { if _, err := runtime.store.SettleCommandSuccess(ctx, store.CommandSuccess{Claim: *claimed.Command, Result: result}, fault.None{}); err != nil { t.Fatalf("SettleCommandSuccess() error = %v", err) } - assertNoNotification(t, listener, 150*time.Millisecond) + waitForNotificationHintKind(t, listener, exec.RunID, store.NotificationEvent, 2*time.Second) } func TestImmediateRetryAndLeaseRecoveryNotify(t *testing.T) { @@ -235,7 +252,7 @@ func TestImmediateRetryAndLeaseRecoveryNotify(t *testing.T) { }, fault.None{}); err != nil { t.Fatalf("SettleCommandConclusion(interrupted) error = %v", err) } - waitForNotificationHint(t, listener, RunID(claim.RunID.String()), 2*time.Second) + waitForNotificationHintKind(t, listener, RunID(claim.RunID.String()), store.NotificationRun, 2*time.Second) candidates, err := runtime.store.ProbeCommands(ctx, []store.CommandKind{{Name: command.Name(), Version: command.Version()}}, 10) if err != nil { @@ -275,7 +292,7 @@ func TestImmediateRetryAndLeaseRecoveryNotify(t *testing.T) { if err != nil || !recovery.Changed { t.Fatalf("RecoverExpiredCommandLease() = %t, %v", recovery.Changed, err) } - waitForNotificationHint(t, listener, RunID(recoveryClaim.RunID.String()), 2*time.Second) + waitForNotificationHintKind(t, listener, RunID(recoveryClaim.RunID.String()), store.NotificationRun, 2*time.Second) } func openNotificationListener(t *testing.T, database testpg.Database, runtime *Runtime) *pgx.Conn { @@ -293,17 +310,19 @@ func openNotificationListener(t *testing.T, database testpg.Database, runtime *R return listener } -func waitForNotificationHint(t *testing.T, listener *pgx.Conn, runID RunID, timeout time.Duration) { +func waitForNotificationHintKind(t *testing.T, listener *pgx.Conn, runID RunID, kind string, timeout time.Duration) { t.Helper() ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - notification, err := listener.WaitForNotification(ctx) - if err != nil { - t.Fatalf("WaitForNotification() error = %v", err) - } - id, valid := store.ParseNotificationHint(notification.Payload) - if !valid || id.String() != string(runID) { - t.Fatalf("notification = %#v, parsed=%s/%t, want run %s", notification, id, valid, runID) + for { + notification, err := listener.WaitForNotification(ctx) + if err != nil { + t.Fatalf("WaitForNotification() error = %v", err) + } + hint, valid := store.ParseNotificationHint(notification.Payload) + if valid && hint.RunID.String() == string(runID) && hint.Kind == kind { + return + } } } @@ -330,7 +349,7 @@ func startAndClaimForNotification( if err != nil { t.Fatalf("Enqueue(%s) error = %v", key, err) } - waitForNotificationHint(t, listener, exec.RunID, 2*time.Second) + waitForNotificationHintKind(t, listener, exec.RunID, store.NotificationRun, 2*time.Second) candidates, err := runtime.store.ProbeCommands(ctx, []store.CommandKind{{Name: command.Name(), Version: command.Version()}}, 10) if err != nil { diff --git a/runtime.go b/runtime.go index feda816..d4198bf 100644 --- a/runtime.go +++ b/runtime.go @@ -132,9 +132,8 @@ func WithPollInterval(interval time.Duration) Option { } // WithNotifications enables or disables transactional PostgreSQL wake hints. -// It defaults to enabled. Polling always remains active and is the correctness -// path, so disabling notifications is suitable for transaction-pooling -// proxies and deliberately poll-only deployments. +// It defaults to enabled. Command scheduling retains polling when disabled, +// but Event.Watch rejects notification-disabled runtimes. func WithNotifications(enabled bool) Option { return runtimeOptionFunc(func(options *runtimeOptions) { options.notifications = enabled }) } @@ -177,6 +176,7 @@ type Runtime struct { runCancel context.CancelFunc runDone chan struct{} wake *wakeHub + eventWakes *eventWakeHub active *activeCommands workerGroup sync.WaitGroup } @@ -223,7 +223,7 @@ func New(db *pgkit.DB, opts ...Option) (*Runtime, error) { poolCapacity: int(db.Conn.Config().MaxConns), observations: observations, faults: options.faults, lifecycle: runtimeCreated, - registry: newRuntimeRegistry(), wake: newWakeHub(), active: newActiveCommands(), + registry: newRuntimeRegistry(), wake: newWakeHub(), eventWakes: newEventWakeHub(), active: newActiveCommands(), }, nil } diff --git a/runtime_run.go b/runtime_run.go index b5e8334..5482d26 100644 --- a/runtime_run.go +++ b/runtime_run.go @@ -359,6 +359,7 @@ func (r *Runtime) Run(ctx context.Context) error { r.lifecycle = runtimeStopping } r.mu.Unlock() + r.eventWakes.close() graceCtx, cancelGrace := context.WithTimeout(context.Background(), r.shutdownGrace) graceful := waitGroupContext(graceCtx, &r.workerGroup) cancelGrace() @@ -384,8 +385,9 @@ func (r *Runtime) Run(ctx context.Context) error { } // runNotificationListener owns exactly one session-capable connection outside -// the application pool. Hints only reduce latency: every connect performs a -// broad catch-up wake, and every scheduler retains its correctness poll. +// the application pool. Command hints only reduce scheduler latency; event +// watches use targeted hints plus a broad catch-up after every connection and +// always re-read durable journal truth. func (r *Runtime) runNotificationListener(ctx context.Context) { const ( initialBackoff = 50 * time.Millisecond @@ -428,6 +430,7 @@ func (r *Runtime) runNotificationListener(ctx context.Context) { // LISTEN begins only after its statement commits. Wake immediately to // close the commit-before-LISTEN and reconnect windows. r.wake.signal() + r.eventWakes.signalAll() r.observe(ctx, Observation{Kind: ObservationRuntime, Operation: "notify_listener", Outcome: "listening"}) for ctx.Err() == nil { if err := r.faults.Hit(ctx, fault.NotifyBeforeWait); err != nil { @@ -437,12 +440,18 @@ func (r *Runtime) runNotificationListener(ctx context.Context) { if waitErr != nil { break } - r.wake.signal() - if _, valid := store.ParseNotificationHint(notification.Payload); valid { + hint, valid := store.ParseNotificationHint(notification.Payload) + if valid { + if hint.Kind == store.NotificationRun { + r.wake.signal() + } + r.eventWakes.signal(hint.RunID) r.observe(ctx, Observation{Kind: ObservationRuntime, Operation: "notify_hint", Outcome: "received"}) } else { // Unknown versions and malformed hints are never interpreted as // work. A bounded broad wake is forward-compatible and safe. + r.wake.signal() + r.eventWakes.signalAll() r.observe(ctx, Observation{Kind: ObservationRuntime, Operation: "notify_hint", Outcome: "broad_wake"}) } } @@ -489,6 +498,9 @@ func (r *Runtime) Stop(ctx context.Context) error { } done := r.runDone r.mu.Unlock() + if r.eventWakes != nil { + r.eventWakes.close() + } if done == nil { return nil } diff --git a/specs/projects/flow/architecture.md b/specs/projects/flow/architecture.md index 156632c..41d8c3e 100644 --- a/specs/projects/flow/architecture.md +++ b/specs/projects/flow/architecture.md @@ -50,7 +50,7 @@ Flow does not replay the full journal on every claim. It also does not treat mut ### 2.4 PostgreSQL is the coordination authority -There is no sidecar broker, distributed lock service, or in-memory leader. PostgreSQL uniqueness, row locks, transactions, `SKIP LOCKED`, durable timestamps, and constraints define accepted state. Process-local channels and `LISTEN/NOTIFY` only reduce latency. +There is no sidecar broker, distributed lock service, or in-memory leader. PostgreSQL uniqueness, row locks, transactions, `SKIP LOCKED`, durable timestamps, and constraints define accepted state. Process-local channels and `LISTEN/NOTIFY` reduce command latency and provide EventWatch liveness; notification payloads never define durable truth. ### 2.5 Durable representation boundaries are explicit @@ -191,8 +191,9 @@ Standalone Flow writes use `READ COMMITTED`. A semantic mutation follows this pr 7. append the immutable batch; 8. update projections, counters, readiness, and queue state; 9. execute an optional fenced application commit callback; -10. issue an optional transactional notification hint only if the transition - created immediately runnable work; and +10. issue bounded transactional run-identity hints when the transition creates + immediately runnable work, records an application event, or terminalizes + the run; and 11. commit or roll back the whole unit. Database time is captured after lock acquisition so transitions serialized on one run also have a consistent decision time. Journal reservation and append occur in the same transaction, so rollback creates no visible gaps. @@ -335,6 +336,14 @@ commands reaching zero are transitioned and queued. The operation reports whether any released command is runnable at database time so notification is limited to useful immediate wakes. +The same journal supports typed future-event inspection without another +durable consumer model. `Event.Watch` registers one run in a process-local +channel hub before capturing `next_journal_position - 1`. Each `Next` snapshots +the run channel before querying the earliest matching post-cursor application +event, which closes the query-to-wait race. Notification startup/reconnect +signals all registered runs; normal valid hints signal only their run. Watches +create no durable rows, timers, worker callbacks, or checked-out connections. + ## 11. Retry and failure transitions Retry policies are canonicalized into every command declaration as opaque bytes with whole-millisecond elapsed/backoff fields. Decisions use PostgreSQL time, persisted budget start, consumed attempts, immutable policy, attempt identity, error classification, and run deadline. Jitter is deterministic and rounded to a durable whole millisecond, so failover replicas calculate the same next time. @@ -378,12 +387,14 @@ There is no global worker-count table. PostgreSQL row locks, queue state, and fe Lease renewals run as one bounded set-oriented statement for the locally active attempts whose individual renewal times are due. Each request carries its own durable duration; a short command never forces unrelated default commands onto its cadence. Exact running fences are selected `FOR UPDATE SKIP LOCKED`, so one row held by settlement cannot block unrelated renewals. Each request is classified as renewed, definitely lost, or uncertain. Definitely lost fences cancel the matching local context immediately; an error or uncertain locked row retains its prior conservative deadline and receives a bounded retry inside that window. A separate earliest-expiry watchdog skips known in-flight renewals, closing the committed-result/local-application race without weakening the durable fence. Both services use shared active-registry timers rather than per-attempt goroutines. Maintenance later recovers expired durable queue rows through the existing path. -Notifications use one separately established session-capable PostgreSQL connection because pool/transaction connections cannot reliably own `LISTEN`. The listener reconnects with bounded backoff and performs a broad wake after every connection to close commit-before-LISTEN gaps. Every scheduler continues polling regardless. +Notifications use one separately established session-capable PostgreSQL connection because pool/transaction connections cannot reliably own `LISTEN`. The listener reconnects with bounded backoff and performs a broad wake after every connection to close commit-before-LISTEN gaps. Every scheduler continues polling regardless; event watches do not poll and therefore require notifications on every writer of their watched runs. -The store emits at most one transactional wake for an operation that creates -immediately runnable work. Claims, journal-only transitions, unmatched events, -terminal settlement without follow-up work, and future-scheduled work do not -notify. +`run` hints wake both the scheduler and run-scoped event watchers. `event` +hints wake only watchers. An application-event operation emits `run` when it +releases immediately runnable work and otherwise emits `event`; run-terminal +journal appends emit `event`. Per-semantic state caps each kind, PostgreSQL +folds identical payloads in a caller transaction, and payloads never contain +event names, keys, or bodies. Claims and future-scheduled work do not notify. ## 14. Runtime lifecycle and deployment @@ -406,7 +417,7 @@ Supported deployment shapes include one all-worker binary, independently scaled ## 15. Inspection, history, and replay -Point/list/queue queries read indexed projections. History reads journal positions directly. Await polls the run projection without reserving a connection between polls. +Point/list/queue queries read indexed projections. History reads journal positions directly. Await polls the run projection without reserving a connection between polls. Event watches read post-baseline journal positions and block on the shared listener without polling or retaining a connection. Trace uses a repeatable-read transaction when it owns the read. It loads bounded history, folds it through the pure replay reducer, loads the live run projection and operational command/wait data in the same snapshot, and overlays those operational fields onto reconstructed semantic commands. A caller-owned Trace inherits the supplied transaction's isolation; callers that require the same coherent cross-statement view must use Repeatable Read or Serializable. diff --git a/specs/projects/flow/benchmark_evidence/plan_16_event_watches.md b/specs/projects/flow/benchmark_evidence/plan_16_event_watches.md new file mode 100644 index 0000000..a981e96 --- /dev/null +++ b/specs/projects/flow/benchmark_evidence/plan_16_event_watches.md @@ -0,0 +1,88 @@ +# Plan 16 event-watch evidence + +Status: Complete + +Measured on 2026-08-14. The baseline was detached at `4550d6e` (the reviewed +Plan 16 text before implementation) with the external-event benchmark's +runtime changed evidence-only from `WithNotifications(false)` to +`WithNotifications(true)`. The after samples used the same command and server +with the Plan 16 implementation in the `plan-16` worktree. + +## Environment + +- Go: `go1.26.5 linux/amd64` +- CPU: Intel Core Ultra 7 255H +- PostgreSQL: 18.1 Debian, x86-64 +- durability: `fsync=on`, `synchronous_commit=on`, `full_page_writes=on` +- application pool: 12 connections + +## Event-only ingress + +Command: + +```sh +go test -run '^$' \ + -bench '^BenchmarkExternalEventIngress/hot_live/no_match$' \ + -benchtime=3s -count=5 . +``` + +| Shape | Baseline range; median | After range; median | Change | +|---|---:|---:|---:| +| latency | 3.927–4.215 ms; 4.139 ms | 4.050–4.463 ms; 4.186 ms | +1.1% | +| rate | 237.3–254.6; 241.6 events/s | 224.1–246.9; 238.9 events/s | -1.1% | +| bytes/op | 22,473–23,106; 22,750 | 22,791–23,751; 23,127 | +1.7% | +| allocs/op | 419 | 429 | +2.4% | + +The event-watch hint adds one transactional `pg_notify` statement to a newly +accepted event-only delivery: the focused query tracer records 8 statements +with notifications disabled and 9 with them enabled, exactly one of which is +`pg_notify`. An event that releases immediately runnable work emits only the +existing stronger `run` hint. Equivalent redelivery, rollback, and a rejected +worker `WithCommit` emit no committed event or hint. Two identical hints issued +for one run by separate semantic operations in one caller transaction are +folded into one delivered notification by PostgreSQL. Median latency remains +well inside the plan's 10% investigation gate. + +The shared external-event fixture now uses the runtime default of notifications +enabled for all of its shapes. Plan 16 repeats and reports only +`hot_live/no_match`, the controlling event-only ingress shape named by the plan. + +## Sparse post-cursor read + +`TestEventWatchSparsePostCursorPlan` runs the exact production query with the +only matching event last. PostgreSQL used the `(run_id, position)` primary-key +range for every shape; no schema/index change was adopted. + +| Post-cursor rows | Execution time | Shared buffers | Rows filtered | +|---:|---:|---:|---:| +| 100 | 0.041 ms | 6 | 99 | +| 1,000 | 0.144 ms | 39 | 999 | +| 10,000 | 1.082 ms | 340 | 9,999 | + +The 10,000-row shape used a bitmap scan of the same primary key and completed +in about one millisecond. This is acceptable for the target run sizes and does +not justify another index. + +## Watch resource shape + +`TestEventWatchThousandIdleWatchersDoNotPoll` constructs 1,000 watches for one +run and starts 1,000 caller-owned `Next` goroutines. After each immediate +durable read completes, an idle interval produces zero additional queries, +the application pool has zero acquired connections, exactly one dedicated +listener connection serves the runtime, command/queue/lease counts are +unchanged, registration itself adds no Flow goroutine, and closing all watches +removes the one shared run entry. + +Focused race coverage also proves targeted unrelated-run isolation, +cross-runtime/multi-watcher broadcast, listener-disconnect catch-up, malformed +hint broad catch-up, disabled-writer characterization, pre-Run listener +startup catch-up, terminal/pruned-run behavior, corruption rejection, +sequential cursor ordering, cancellation reuse, live-run replacement, and +runtime shutdown cleanup. Worker-staged events wake a remote watch only after +successful settlement. `Close` also cancels a `Next` blocked on application- +pool acquisition, and stopping a runtime cancels watch construction blocked on +its initial database read. + +Final verification used a reset PostgreSQL database. The complete ordinary and +race suites, build, vet, formatting, module tidy/verification, and diff checks +passed. The named-test audit ran 531 tests with zero named skips or failures. diff --git a/specs/projects/flow/components/engine.md b/specs/projects/flow/components/engine.md index de54cf7..ff2173f 100644 --- a/specs/projects/flow/components/engine.md +++ b/specs/projects/flow/components/engine.md @@ -13,6 +13,10 @@ The engine owns typed contracts and deterministic worker decisions. It transform Commands retain name/version, argument/result codecs, retry policy, attempt timeout, and queue. Events retain a name and payload codec. `Event.Deliver` is deliberately detached targeted ingress to a known run, including from application code inside a worker attempt. Definitions are immutable; invalid names/versions/options and nil workers fail validation. +`Event.Watch` reuses that same typed definition for read-side inspection of +future events in one known run. It does not add an event handler, subscription, +callback, acknowledgement, or executable engine concept. + `Work[A]` is the attempt-local scope for one claimed command, not the whole run or the immutable command definition. A fresh value is created for every worker invocation. It exposes typed `Args` and immutable `CommandInfo`, diff --git a/specs/projects/flow/components/runtime.md b/specs/projects/flow/components/runtime.md index c65de84..a9b54e1 100644 --- a/specs/projects/flow/components/runtime.md +++ b/specs/projects/flow/components/runtime.md @@ -40,6 +40,13 @@ later application writes under caller commit ownership. A new event appends history, resolves exact wait rows, and updates command readiness in one transaction. Worker-staged events use the same transition during settlement. +`Event.Watch` is a typed read-side view over future application-event journal +rows. It registers one run in a shared local channel hub, captures the journal +head, and performs a durable read before blocking the caller's `Next` +goroutine. Valid hints target one run; listener startup/reconnect and malformed +hints conservatively signal all current registrations. Watches allocate no +goroutine, timer, connection, command, lease, or acknowledgement of their own. + `ReplaceCurrentRun` locks the exact expected live-key predecessor, cancels it, and inserts a distinct successor in one transaction. An unexpected equivalent current ID is only rediscovered for retry/ambiguous-commit recovery; a different @@ -52,7 +59,10 @@ The maintenance scheduler expires unresolved wait budgets independently of initi Active attempts retain IDs, tokens, their resolved lease duration, conservative local expiry, next renewal, in-flight/retry state, and cancellation function. One earliest-due service renews all due attempts in one mixed-duration, time-bounded statement. Exact fences use `FOR UPDATE SKIP LOCKED` and classify every request as renewed, lost, or uncertain. Errors and uncertain rows receive bounded retries inside their remaining local window; a retry may use more than the ordinary five-second cap but never an unbounded database call. Lost cancels only the matching attempt. The independent earliest-expiry watchdog ignores a matching renewal while its result is in flight, then cancels locally expired contexts. Both services share one lightweight registry-change signal, not a goroutine or timer per attempt. Maintenance recovers expired durable leases unchanged so another replica can retry safely. -Notification hints reduce wake latency. Polling remains sufficient through transaction-pooling proxies, lost messages, reconnects, or disabled notifications. +Notification hints reduce command wake latency, whose scheduler retains +polling. Event watches intentionally have no polling repair path: they require +notifications enabled on every writer and use listener startup/reconnect +catch-up plus caller-bounded contexts. Cancellation stops claims/listening/maintenance, waits through shutdown grace, then cancels remaining worker contexts. The caller-owned database pool is never closed. diff --git a/specs/projects/flow/components/schema.md b/specs/projects/flow/components/schema.md index b2d46f6..3628ae1 100644 --- a/specs/projects/flow/components/schema.md +++ b/specs/projects/flow/components/schema.md @@ -104,9 +104,12 @@ failure transition. External event ingress through `Event.Deliver` enforces exact target-local identity, appends the event, records satisfying positions, and applies delta readiness. Equivalent repeats are idempotent; conflicting repeats fail; terminal runs cannot be reopened. Delivery adds no source identity or storage shape and uses a caller transaction unchanged when supplied. -The generic journal append emits no scheduler notification. A semantic -operation may emit at most one transactional hint after it creates work that is -immediately runnable at database time; polling remains the correctness path. +The generic journal append emits no scheduler wake for journal-only work. An +application-event operation emits one transactional `run` hint when it creates +immediately runnable work and otherwise emits an `event` hint; run-terminal +entries emit `event`. Both contain run identity only. Command scheduling +retains polling; event watches rely on the shared listener and startup/ +reconnect catch-up. Bounded indexed maintenance recovers command leases, expires unresolved waits, and enforces run deadlines. Inspection uses indexed lookup/keyset pagination. Trace folds the journal under repeatable read and overlays bounded operational command data. diff --git a/specs/projects/flow/functional_spec.md b/specs/projects/flow/functional_spec.md index c6b72f8..2cfe453 100644 --- a/specs/projects/flow/functional_spec.md +++ b/specs/projects/flow/functional_spec.md @@ -273,6 +273,33 @@ part of ordinary event readiness resolution. Retries and lease takeovers receive the same immutable payloads selected by the recorded satisfying positions. +### 6.3 Future-event inspection + +`event.Watch(ctx, runtime, runID)` captures the current journal head and +returns an `EventWatch[T]` for later matching application events. A watch is a +broadcast read: it does not consume or acknowledge an event, create a command, +hold a connection, invoke a callback, or count against worker concurrency. +`Next(ctx)` returns `(eventKey, typedPayload, error)` in journal order and must +be called sequentially. `Close` is idempotent and releases the local +registration. + +Applications establish the watch before reading their own projection. An +event committed around that boundary is then either included in the watch +baseline and visible to the application read, or appears after the baseline +and is returned by `Next`. Application tables remain the response authority. +If the run becomes terminal, matching post-baseline events are drained first; +otherwise `Next` returns `ErrTerminal` so a live-key caller can re-read and +resolve a replacement generation. A retained run pruned after construction is +also terminal to the watch. + +`Next` performs one immediate durable journal read and then waits only for a +run-scoped PostgreSQL hint, listener startup/reconnect catch-up, caller +cancellation, `Close`, or runtime shutdown. It has no periodic poll or +internal deadline. Callers must use a context matching their latency contract. +Every runtime that writes a watched run through `Deliver`, `Emit`, settlement, +cancellation, expiry, or replacement must have notifications enabled; +notification-disabled runtimes cannot create watches. + ## 7. Successful settlement and `WithCommit` After a worker returns successfully, Flow validates the attempt context, canonicalizes the result, normalizes the decision, and reacquires the run under the attempt fence. @@ -479,12 +506,17 @@ Public options are: - one reconnecting PostgreSQL notification listener when enabled; and - asynchronous observer delivery. -Polling is the correctness path. Notifications are transactional latency hints; malformed/lost hints, listener disconnects, transaction-pooling proxies, and disabled notifications do not lose work. +Polling remains the command-scheduler and maintenance correctness path. +`EventWatch` deliberately does not poll: it requires notifications, performs a +durable read after targeted hints, and catches disconnected commits when the +listener successfully starts or reconnects. A prolonged listener outage is +bounded by the caller's `Next` context. -Flow emits a wake hint only when a committed transition creates work that is -immediately runnable. Journal-only transitions, claims, terminal settlements -without follow-up work, unmatched events, and work scheduled for the future do -not require a hint. +Flow emits a `run` hint when a committed transition creates immediately +runnable work. It also emits an `event` hint for durable application events and +run terminal transitions when no stronger hint has already covered the +semantic operation. Both payloads contain only version, kind, and run ID; +claims and future-scheduled work do not notify. The scheduler may claim selected groups from independent runs concurrently using an internal bound derived from worker capacity and the @@ -566,6 +598,10 @@ caller transaction and its uncommitted writes. `AwaitRun` polls without holding a worker, lease, or database connection between reads until the run is terminal or the context ends. +`EventWatch` instead observes future events for one known run. It uses the +runtime's shared listener, performs no periodic read while idle, and requires +notifications enabled on all writers of that run. + `GetQueueStats(ctx, client, queues...)` accepts at most 200 queue-name inputs before deduplication and returns an entry for every requested distinct lane, including empty lanes. Ready, delayed, running, and oldest-ready values use one SQL diff --git a/specs/projects/flow/implementation_plan.md b/specs/projects/flow/implementation_plan.md index 42f946b..82edd38 100644 --- a/specs/projects/flow/implementation_plan.md +++ b/specs/projects/flow/implementation_plan.md @@ -24,6 +24,10 @@ completed in PR #27 at `a98256d` and was released as v0.4.2. completed its accepted streamlined scope in PR #30 at `b25fbe3` and was released as v0.4.3; the larger unused API/registry/example proposal was closed rather than retained as required work. +[`plans/16-distributed-application-event-watches.md`](plans/16-distributed-application-event-watches.md) +is implemented and awaiting its reviewed release commit. It adds typed, +run-scoped future-event inspection through the existing PostgreSQL listener, +without polling or adding a durable subscription model. ## Implemented Plans 9–10 outcomes diff --git a/specs/projects/flow/plans/16-distributed-application-event-watches.md b/specs/projects/flow/plans/16-distributed-application-event-watches.md new file mode 100644 index 0000000..076b0ba --- /dev/null +++ b/specs/projects/flow/plans/16-distributed-application-event-watches.md @@ -0,0 +1,827 @@ +# Plan 16: Await durable application events across runtime instances + +Status: Implemented; release pending + +> **Executor instructions:** Read this plan completely before editing. Follow +> the phases in order and run every phase's verification before continuing. +> This is a read-side observation feature, not a second workflow engine: +> watches must never create commands, leases, journal acknowledgements, or +> callback workers. PostgreSQL remains the durable authority and +> `LISTEN/NOTIFY` is an identity-only wake signal, never the returned truth. +> Stop and report any design that requires a connection, periodic timer/query, +> additional background goroutine, or durable row per waiting caller. `Next` +> may block the caller's existing goroutine under its supplied context. +> +> **Drift check (run first):** +> +> ```sh +> git status --short --branch +> git diff --stat d7277f9..HEAD -- \ +> '*.go' internal/store README.md specs/projects/flow +> ``` +> +> Re-read the current event-ingress, journal, notification, and runtime +> lifecycle code named in Section 2 if any in-scope source changed after +> `d7277f9`. A change to application-event identity, the run-first lock order, +> notification payloads, journal retention, or runtime shutdown is a STOP +> condition until this plan is updated. + +## 1. Status and dependency + +- **Priority:** P1; required by Trails API plan 012 +- **Effort:** L +- **Risk:** MEDIUM; the public API is additive, but notification routing and + missed-wake behavior are distributed-concurrency code +- **Depends on:** Plans 13, 14, and 15 completed on `master` +- **Category:** runtime / read API / distributed coordination / performance +- **Planned at:** `d7277f9a27e0871c8e5bb74b1ed56f546b9e2a1a` + (`master`, one documentation commit after `v0.4.3`), 2026-08-13 +- **Consumer:** Trails API plan 012, whose `WaitIntentReceipt` handler needs to + await an `intent.receipt.changed` fact for the current `intent.run` +- **Recommended release:** `v0.5.0`, because this adds a public capability; + Trails must consume the actual tag produced after this plan passes + +## 2. Why this matters + +Flow already owns the hard parts of a durable event: immutable run-local +identity, atomic journal persistence, cross-replica PostgreSQL coordination, +and a dedicated notification connection. It currently exposes application +events only as inputs to durable commands. An HTTP handler that wants to wait +for the *next* application fact cannot reuse that machinery without creating a +fake command or independently rebuilding a database listener and poll loop. + +That gap is causing the first production embedder to duplicate Flow in the +application: Trails PR 1063 adds application triggers, another `LISTEN` +connection, a process-local subscriber map, server restart policy, and a +second correctness poll solely to wake `WaitIntentReceipt`. The desired model +is smaller: + +```text +application transaction + -> records an immutable Flow application event + -> commit emits a tiny run-scoped hint + -> every Flow runtime listening to that database receives the hint + -> local waiters for that run re-read the durable Flow journal + -> application re-reads its own projection and returns if it changed +``` + +The event payload in PostgreSQL's notification is not the application's source +of truth: the durable journal is. `LISTEN/NOTIFY` wakes the read, and a +successful listener start/reconnect signals every registered watch so events +committed during a connection gap are discovered. There is no periodic watch +poll. Callers bound `Next` with their own context and re-read application truth +when that context ends. No sticky sessions or application-specific database +triggers are required. + +## 3. Current state and verified evidence + +The implementation starts from these facts at the planned commit: + +- `definitions.go:23-26` defines typed `Event[T]`; `DefineEvent` fixes its + application namespace and codec. +- `enqueue.go:329-397` implements `Event.Deliver`. Identity is immutable by + `(run ID, event name, event key)`, and an equivalent repeat is idempotent. + A different payload for the same identity is `ErrConflict`. +- `flow.Emit` stages same-run events atomically with a successful worker + decision. `Event.Deliver` can join a caller-owned transaction through + `Runtime.InTx`. +- `flow_journal` is ordered by `(run_id, position)` and already stores event + name, key, class, canonical body, and recorded time. `flow_runs` already + carries `next_journal_position`; no new cursor table is needed. +- `types.go:18` declares `JournalPosition`; `history.go` exposes the durable + history projection. Using general `History` would force every caller to + decode/filter pages and invent its own wait protocol. +- `inspection.go:238-269` implements `AwaitRun` with a durable row check, a + process-local broad wake, and its existing timer fallback. That separate API + waits only for terminal run state; `Event.Watch` neither reuses its timer nor + reports command/run completion as an application event. +- `runtime_run.go:389-458` owns one session-capable PostgreSQL connection per + runtime, outside the application pool. PostgreSQL sends each committed + notification to every listening session, which is the distribution model + required by handlers arriving on arbitrary pods. +- The listener currently parses a run ID and then calls only the broad + scheduler `wakeHub`; it discards the parsed identity. +- `internal/store/store.go:302-327` emits a notification only when a transition + creates immediately runnable commands. An application event with no durable + command gate therefore records correctly but sends no hint. +- The notification payload is deliberately small and versioned: + `{"v":1,"kind":"run","key":""}`. Unknown payloads cause a safe + broad wake. +- Plans 13 and 14 retain polling for command scheduling and maintenance. This + plan does not change that engine contract and does not reuse its poll for + read-side event watches. A watch progresses through committed notification + hints, listener start/reconnect catch-up, or caller cancellation only. + +The functional specification currently says Flow omits event handlers and +command-outcome subscriptions. Preserve that boundary. A caller awaiting an +immutable application event is an inspection operation: it does not consume +the event, run a callback, or durably react to it. + +## 4. Controlling design + +### 4.1 Public API + +Add a typed future-event watch to `Event[T]`: + +```go +func (event Event[T]) Watch( + ctx context.Context, + runtime *Runtime, + runID RunID, +) (*EventWatch[T], error) + +func (watch *EventWatch[T]) Next(ctx context.Context) ( + key string, + payload T, + err error, +) +func (watch *EventWatch[T]) Close() +``` + +The names and signatures above are the complete target API. `EventWatch` is the +only new exported type and has unexported state. `Next` returns the immutable +application event key and typed payload; the caller already knows the run and +event definition. Keep the journal cursor, position, event ID, recorded time, +and listener state private. Do not add an exported event-record/cursor type, +watch options, a Trails-specific helper, or access to internal store/journal +types. + +Contract: + +1. `Watch` validates the event definition and run ID locally, registers for + run-scoped wake hints, then validates the durable run/status and captures + the current journal head as its baseline cursor in one store read. Events + at or before that cursor are historical and are not returned. +2. Register-before-cursor-read is required. If an event commits during watch + construction, the cursor includes it and the application read that follows + sees the transaction that produced it. +3. `Next` snapshots the current targeted-hub generation/channel *before* it + queries durable journal state for the earliest matching application event + with `position > cursor`. If the query finds neither an event nor a terminal + run, it waits on that channel, context cancellation, caller `Close`, or + runtime shutdown. It repeats the durable query after every hint/startup/ + reconnect wake. Taking the wait channel before the query closes the local + query-to-wait race without a timer. +4. A returned event advances the cursor to its position. Sequential `Next` + calls drain matching events in journal order. Concurrent `Next` calls on one + watch are invalid; document and reject them rather than introducing + nondeterministic cursor ownership. +5. `Close` is idempotent, removes the local registration, and unblocks an + active `Next` with `ErrClosed`. Context cancellation returns `ctx.Err()` + but does not implicitly close the watch; a caller may use another context + for a later `Next` and remains responsible for `Close`. +6. A missing run at watch creation is `ErrNotFound`. A run that is already + terminal at watch creation is `ErrTerminal`. Both are ordinary races for a + caller that resolved a live key before calling `Watch`: that caller should + re-read application truth and, if needed, resolve the current run again. +7. If the run becomes terminal, `Next` returns any matching event already + committed after the cursor first; otherwise it returns `ErrTerminal`. + This lets a caller re-read application truth and, for live-key replacement, + resolve the new current run. If the previously verified run disappears + because an explicit retention pass pruned it after terminal settlement, + `Next` also returns `ErrTerminal`; a watch never pins retained history. +8. `Watch` accepts `*Runtime`, not `Client`. The concrete type encodes that a + watch needs the runtime's process-local listener/hub and makes a transaction + client impossible to pass. A nil runtime is `ErrInvalid`. +9. The watch holds no PostgreSQL connection between reads, creates no durable + row, command, wait, lease, or acknowledgement, and does not count against + worker or queue concurrency. +10. Watchers are broadcast readers. Any number of runtimes and callers may see + the same event; nothing is consumed globally. +11. `Watch` requires notifications to be enabled. A runtime configured with + `WithNotifications(false)` is rejected as `ErrInvalid`. A created runtime + may register a watch before `Runtime.Run`; it receives no future-event wake + until `Run` successfully establishes `LISTEN`, whose startup catch-up signal + then forces a durable read. The caller context bounds a runtime that never + starts. A stopping/stopped runtime is `ErrClosed`, and stopping an accepted + runtime closes all active watches. + +The documented race-free application pattern is: + +```go +watch, err := changed.Watch(ctx, runtime, runID) +if errors.Is(err, flow.ErrTerminal) || errors.Is(err, flow.ErrNotFound) { + // The run settled, was replaced, or was pruned after lookup. Application + // state is still the response authority. + return readApplicationProjection(ctx) +} +if err != nil { + return err +} +defer watch.Close() + +projection, err := readApplicationProjection(ctx) +if err != nil || projectionIsReady(projection) { + return err +} + +for { + if _, _, err := watch.Next(ctx); errors.Is(err, flow.ErrTerminal) { + return readApplicationProjection(ctx) + } else if err != nil { + return err + } + projection, err = readApplicationProjection(ctx) + if err != nil || projectionIsReady(projection) { + return err + } +} +``` + +The watch must be established before the application read. A committed event +is then either included in the watch baseline and visible to the subsequent +application read, or it is after the baseline and returned by `Next`. A +terminal error from either watch construction or `Next` is a reason to re-read +application truth, not a reason to return stale projection data. + +`Next` deliberately has no internal deadline. A listener outage that has not +yet reconnected can therefore hold it until the caller's context ends. Public +documentation and examples must always use a context appropriate to the +consumer's latency contract. This is preferable to multiplying periodic +database reads by the number of waiters. + +The notification requirement applies to producers as well as consumers. Every +runtime that may `Deliver`, `Emit`, terminalize, or replace a run observed by +`Event.Watch` must have notifications enabled. A notification-disabled writer +can still record durable state for Flow's existing poll-only command mode, but +it cannot wake a watch and there is deliberately no timer repair. Flow cannot +enforce another process's configuration through local API validation, so this +is a documented deployment invariant for applications adopting watches. + +### 4.2 Durable query contract + +Add narrow store reads rather than implementing watches through public +`History` pages: + +- capture the current run head and `next_journal_position - 1` in one query; +- read the earliest application event for one `(run ID, event name)` after one + position; and +- read run terminal status in the same query, so an idle watch does not need an + additional `GetRun` round trip. + +The event query is conceptually one statement with one lateral event lookup: + +```sql +SELECT r.status, + next_event.position, next_event.event_key, next_event.body +FROM flow_runs AS r +LEFT JOIN LATERAL ( + SELECT position, event_key, body + FROM flow_journal + WHERE run_id = r.run_id + AND position > $2 + AND event_namespace = 'application' + AND event_class = 'application' + AND event_name = $3 + ORDER BY position + LIMIT 1 +) AS next_event ON true +WHERE r.run_id = $1; +``` + +Interpret the statement in one fixed order: return a non-null matching event +first even when the run is terminal; if no event exists and the status is +terminal, return `ErrTerminal`; otherwise wait. No row means `ErrNotFound` +during construction and `ErrTerminal` after construction already verified the +run. + +Decode through the event definition's existing codec and canonical +application-event envelope. A malformed retained body is `ErrInvalidState`. +Advance the private cursor with `position`, but return only `event_key` and the +typed payload. Never return raw journal bodies or export journal metadata from +this API. + +The primary key `(run_id, position)` starts the scan at the watch cursor, so +old history before watch creation is not revisited. A running run has no strict +journal-entry ceiling, however, so do not assume every run is small. Do not +change the published baseline migration or add a schema migration by default. +Before considering an index, use `EXPLAIN (ANALYZE, BUFFERS)` at 100, 1,000, +and 10,000 post-cursor entries with a sparse matching event at the end. Add a +follow-up migration only if measured work is materially unbounded for the +target workload; that is a STOP-and-report decision, not pre-approved scope +for this plan. + +### 4.3 Targeted local wake hub + +Add a runtime-owned event wake hub keyed by parsed `RunID`. It should use a +generation/channel pattern like the existing `wakeHub`, with these additions: + +- multiple watchers for one run each wake on one signal; +- a signal is coalesced, not queued per event; +- registration/unregistration is bounded and leak-free; +- successful initial `LISTEN` and every reconnect perform a catch-up signal for + all current watches; +- runtime shutdown closes all watches; and +- an unrelated run hint does not wake or re-query this run's watches. + +Use one shared generation/channel entry per watched run plus one close channel +per `EventWatch`; do not allocate one hub goroutine or one notification queue +per watcher. Remove the run entry when its final watch closes. Watch +construction must unregister on every validation/cursor-read failure. + +Each `Next` loop obtains the hub's current generation/channel before issuing +its durable read. A signal between that read and the blocking select therefore +leaves the captured channel closed and forces another durable read. Do not use +a timer to cover an incorrectly ordered query/wait sequence. + +Keying only by run ID is intentional. PostgreSQL hints do not carry payloads or +event names; durable queries perform the exact event-name filter. This keeps +notification payloads small and makes a single hint cover several application +events committed for the same run. + +### 4.4 Notification hint vocabulary + +Retain version 1 and the existing `kind:"run"` payload for immediately +runnable work. Add one understood kind, for example: + +```json +{"v":1,"kind":"event","key":""} +``` + +Semantics: + +- `run`: wake the command scheduler and event watches for that run; +- `event`: wake only event watches for that run; +- unknown/malformed: retain the conservative broad scheduler wake and signal + all event watches, because no run identity can safely be trusted. + +Keep payload version 1. During a rolling upgrade, an older runtime treats the +new `event` kind as unknown and performs its existing conservative scheduler +wake; a newer runtime routes it to watchers. Hints contain no durable truth, so +no compatibility decoder or second payload version is needed. + +Emit an `event` hint when a transaction records an application event but does +not make a command runnable. If that same semantic transaction makes work +runnable, a `run` hint is sufficient for both consumers. + +Keep notification state local to the existing run-scoped `SemanticTx` and its +`notificationOwner`; do not add a transaction registry, transaction callback, +or map of touched runs. The simple rules are: + +- suppress an `event` hint if that semantic operation already requested an + `event` or stronger `run` hint; +- suppress a repeated `run` hint; +- if an `event` hint was already requested and the same transaction later + makes work runnable, permit one `run` hint as the upgrade; and +- let PostgreSQL fold identical channel/payload notifications produced by + separate semantic operations in one caller-owned transaction. + +The ordinary maximum is therefore one `event` plus one `run` payload for one +run in one transaction, not an elaborate application-side exactly-one +protocol. Notifications are identity-only wake signals and distinct runs have +distinct payloads. The durable journal, not notification delivery or payload, +is the event authority. A hint for run A must never suppress a hint for run B. +Do not send event payloads, names, keys, tenant identifiers, or secrets through +`pg_notify`. + +Implement this by replacing the current single `notificationSent` bit with the +smallest equivalent per-semantic state, such as `eventHintSent` and +`runHintSent`, inherited through `notificationOwner`. Do not introduce a +general-purpose notification abstraction. Keep `NotifyRunnableCommands` for +the stronger existing `run` hint and add one narrowly named internal method +for the `event` hint, such as `NotifyEventWatchers`. + +Run terminal transitions must also wake event watches. A watcher awaiting an +event that will never occur needs prompt `ErrTerminal`, especially when +`ReplaceCurrentRun` cancels one live-key holder and creates another. Emit a +run-scoped event-watch hint for terminal settlement even when no command became +runnable. Replacement must wake watchers for the predecessor; the caller then +re-resolves the live key and watches the replacement. + +Call the narrow event-hint method from the store operation that already knows +an application event was accepted without immediate readiness, or that a run +projection became terminal. Do not emit from observers, public API wrappers, +or a duplicated list of after-commit call sites; those locations either run +too late or cannot share caller-owned transaction atomicity. + +All notification calls remain inside the semantic transaction. PostgreSQL +must deliver them only if that transaction commits. A rolled-back application +event or terminal transition must produce neither durable data nor an +actionable wake. Do not add a general after-commit hook. A same-runtime caller +receives its own committed PostgreSQL notification when notifications are +enabled. Event watches reject notification-disabled runtimes; command +scheduling may continue to support its existing poll-only deployment mode. A +local post-commit signal is allowed only at an existing code boundary that +already knows a runtime-owned commit succeeded, and is not required for this +plan. + +### 4.5 Notification-only waiting and efficiency + +There is no event-watch poll interval, fallback mode, or timer-driven query. +One `Watch` construction reads its baseline. One `Next` call reads immediately, +then reads again only after a targeted notification, listener startup/reconnect +catch-up, or another explicit hub signal. Context expiry returns `ctx.Err()` +without another Flow query; the application decides whether its response +contract requires one final projection read. + +The listener's successful `LISTEN` is the recovery boundary. It signals every +registered watch before waiting for notifications, closing both the initial +commit-before-listen window and every disconnected interval. If the listener +cannot reconnect, bounded callers time out normally; Flow must expose the +existing `listening`, `connect_error`, and `reconnecting` observations so +operators can diagnose the outage. Emit state changes, not a log/metric per +watch. Tests must prove a connect failure and successful recovery produce the +expected low-cardinality lifecycle observations. Do not claim an unbounded +liveness guarantee when PostgreSQL is unavailable. + +Do not create a Flow-owned goroutine or timer per registered watch while it is +idle. `Next` blocks only the caller's existing goroutine on channels/context. +There is one dedicated PostgreSQL listener connection per runtime, not per +watch, and no application-pool connection remains checked out. Multiple +waiters may each perform the durable read needed to decode their result after a +hint; do not add a cache of typed payloads in this phase. + +## 5. Scope + +### In scope + +- `event_watch.go` and `event_watch_test.go` (new) for the typed public watch + and its state. +- `runtime.go`, `runtime_run.go`, and focused tests for the run-targeted wake + hub, listener routing, reconnect, and shutdown. +- `internal/store/store.go` plus a narrow store read file/test for cursor, + next-event, terminal, and notification decisions. +- `enqueue.go`, `command_runtime.go`, and existing `internal/store` event and + terminal transition paths required to emit a committed hint. Touch only + paths that already know an event was accepted, work became runnable, or a + run became terminal. +- `README.md`, `specs/projects/flow/functional_spec.md`, + `specs/projects/flow/architecture.md`, and the relevant runtime/engine + component specs. +- Compile-contract coverage for the exported generic API. + +### Out of scope + +- Persistent subscriptions, callbacks, event handlers, webhooks, consumers, + acknowledgement offsets, or global topics. +- Waiting for arbitrary command outcomes or OR/quorum/race conditions. +- A durable command whose only work is waiting for another event. +- Searching runs by untyped “entity ID.” Applications use their root command + definition and stable run key with `GetCurrentRun`. +- Direct application access to Flow SQL or the notification channel. +- New tables, columns, triggers, brokers, advisory locks, or a required schema + migration. +- Changing application-event identity, `WaitFor` gate semantics, journal + retention, run lock order, attempt fencing, queue scheduling, or delivery + guarantees. +- A watch that works with notifications disabled; that configuration has no + event-wake owner without polling. +- Cross-process configuration discovery or enforcement for notification- + disabled writers; watch-using deployments must keep notifications enabled on + every writer of the watched runs. +- Any event-watch poll/fallback option, shared typed-payload cache, + transaction-wide notification registry, or general after-commit callback + mechanism. + +## 6. Implementation phases + +### Phase 0 — characterize the current contract + +Before editing, add or identify tests proving: + +- an application event with no matching command wait records durably but emits + no current notification; +- an application event that releases a wait emits the existing run hint; +- each runtime instance connected to the same database receives committed + hints; and +- a rollback emits no visible notification. + +Record the current query plan for the Section 4.2 shape at 100, 1,000, and +10,000 post-cursor journal entries with the matching event last. Also retain a +five-sample baseline for the existing no-wait external-event ingress benchmark +with notifications enabled. This is evidence only; do not add an index unless +the STOP gate is reached. + +**Verify:** + +```sh +go test -count=1 -p 1 -parallel 4 -run 'Notification|Deliver|Event' ./... +go test -run '^$' \ + -bench '^BenchmarkExternalEventIngress/hot_live/no_match$' \ + -benchtime=3s -count=5 ./... +``` + +Expected: all characterization tests pass and the no-waiter event test proves +the exact missing hint this plan addresses; five benchmark samples and the +three query plans are recorded as baseline evidence. + +### Phase 1 — add typed durable event reads + +Implement the store cursor/next-event query and payload decoding. Add the +public `EventWatch[T]` handle and the exact `Watch`/`Next`/`Close` API from +Section 4.1 with direct durable reads plus context/close behavior first. Reject +invalid events, nil runtimes, invalid/missing/terminal runs, +notification-disabled runtimes, concurrent `Next`, and use after close with +existing Flow sentinel categories. Map disappearance of a run that was +verified during watch construction to `ErrTerminal`; do not pin or recreate +pruned history. Use a directly signaled internal hub in tests until Phase 2 +routes database hints; do not add a timer. + +Tests must cover: + +- `compile_contract_test.go` pins `Event[T].Watch` to + `func(context.Context, *Runtime, RunID) (*EventWatch[T], error)` and `Next` + to `func(context.Context) (string, T, error)` so later work cannot widen the + surface back to `Client` or an exported record without an explicit API + decision; +- events before the watch baseline are not returned; +- several events after the cursor return in position order; +- other event names and runtime events are ignored; +- event keys and typed payloads round-trip without exposing run ID, journal + position, event ID, or recorded time through a new public record type; +- malformed stored payload returns `ErrInvalidState`; +- a matching event committed between the pre-wait query and the channel wait + is found on the next durable read; +- terminal-with-event returns the event before `ErrTerminal`; +- missing and terminal-at-start construction races; +- terminal pruning after construction returns `ErrTerminal`; +- nil-runtime, close, and concurrent-`Next` behavior; +- a cancelled `Next` returns `ctx.Err()`, remains reusable, and is removed only + when the caller invokes `Close`; +- `WithNotifications(false)` returns `ErrInvalid`, a created runtime permits a + bounded watch that gains wakeups when `Run` starts, and a stopped runtime + returns `ErrClosed`; and +- after its immediate read, an unchanged `Next` performs no second query before + an explicit hub signal or context cancellation. + +**Verify:** + +```sh +go test -race -count=1 -p 1 -parallel 4 -run 'EventWatch' ./... +``` + +Expected: all new API and durable-read tests pass with no race report. + +### Phase 2 — route targeted hints across runtimes + +Add the run-targeted event wake hub and route parsed notification payloads. +Preserve the existing broad scheduler wake for `kind:"run"`. Add reconnect +catch-up and shutdown behavior. Do not let watcher map locks cover a database +query or caller callback. These routing tests may commit `pg_notify` directly +after arranging durable event state; Phase 3, not this phase, connects normal +event/terminal writes to the new hint. + +Tests must start two runtimes against one database and prove: + +- a committed test `pg_notify` carrying `kind:"event"` for a watched run wakes + Runtime B within a bounded test deadline; +- two runtimes and multiple watchers on each all observe the same fact; +- an unrelated run hint does not wake/query the watched run; +- a malformed/future hint performs a conservative catch-up; +- forced listener disconnect/reconnect cannot strand a committed event; +- an event committed while the listener is disconnected is returned after the + reconnect catch-up signal, even when no later event commits; +- a prolonged injected listener outage lets `Next` reach its caller deadline + without periodic queries, and the same watch remains reusable after reconnect; +- caller `Close` and `Runtime.Run` shutdown remove every registration; +- notification-disabled runtimes reject `Watch`; a pre-`Run` watch whose event + is already durable wakes from the listener's initial catch-up when `Run` + starts; and +- a notification-disabled producer does not wake a watch on another runtime, + documenting why watch-using deployments require every writer to enable + notifications; and +- no application-pool connection remains checked out while `Next` waits. + +Also assert the shared observer receives `connect_error`/`reconnecting` and +the later `listening` recovery outcome without any per-watch observation storm. + +**Verify:** + +```sh +go test -race -count=1 -p 1 -parallel 4 -run 'EventWatch|NotificationListener' ./... +``` + +Expected: cross-runtime and lifecycle tests pass with no race report or leaked +connection/registration. + +### Phase 3 — emit event and terminal hints transactionally + +Generalize notification parsing/state with the small per-semantic rules from +Section 4.4: + +- application event plus immediate readiness -> `run`; +- application event without readiness -> `event`; +- terminal run without readiness -> `event`; +- no externally relevant change -> no new hint. + +Exercise `Event.Deliver`, staged `Emit`, worker settlement, direct cancellation, +deadline expiry, and `ReplaceCurrentRun`. Keep existing runnable-command hint +behavior and ambiguous-commit/fault-hook boundaries unchanged. + +Tests must prove: + +- event-only commits now wake remote watches without waking the new runtime's + scheduler path; +- Runtime A commits `Event.Deliver`; a watch created through Runtime B wakes + within a bounded deadline and returns the durable typed event; +- application event plus released command emits no more than one `event` and + one `run` hint for that run and both scheduler and watcher progress; +- identical payloads from repeated semantic operations in one caller-owned + transaction are folded by real PostgreSQL without a Flow transaction map; +- equivalent idempotent event redelivery does not generate another semantic + event/hint; +- rollback and failed `WithCommit` generate no durable event/hint; +- every terminal status wakes the watch to `ErrTerminal`; and +- atomic live-key replacement wakes the predecessor watch and exposes the new + holder through `GetCurrentRun`. + +**Verify:** + +```sh +go test -race -count=1 -p 1 -parallel 4 -run 'EventWatch|Notification|ReplaceCurrentRun|Terminal' ./... +``` + +Expected: all notification, replacement, and terminal regressions pass. + +### Phase 4 — document and measure the contract + +Update public and project documentation with: + +- the watch-before-application-read recipe; +- the distinction among `WaitFor`, `Event.Deliver`, `Event.Watch`, `History`, + `GetResult`, and `AwaitRun`; +- cross-runtime broadcast behavior and lack of sticky-session requirements; +- notification wake, reconnect catch-up, caller-context, and no-poll semantics; +- listener lifecycle observations operators should alert on; +- terminal/replacement behavior; and +- the rule that application tables remain source of truth. + +Add a benchmark or bounded query-count test for 1,000 idle watches on one +runtime. It need not assert wall-clock speed. It must prove one listener +connection, no worker/queue/lease growth, no goroutine created merely by +registration, no application connection held while waiting, and no query-count +growth during an idle observation interval after each caller's initial `Next` +read. Test goroutines that invoke `Next` represent caller goroutines and must +all exit through context cancellation; Flow must not create another goroutine +for them. + +Repeat the existing no-wait external-event ingress benchmark for five samples +with notifications enabled. Record median/range and PostgreSQL protocol/query +count before and after. Investigate before completion if the new hint adds more +than 10% median latency on the same environment or introduces work that scales +with retained history. Small variance below that threshold is evidence, not a +reason to add batching, caching, or another abstraction. + +**Verify:** + +```sh +go test -count=1 -p 1 -parallel 4 ./... +go test -run '^$' \ + -bench '^BenchmarkExternalEventIngress/hot_live/no_match$' \ + -benchtime=3s -count=5 ./... +test -z "$(gofmt -l .)" +go vet ./... +go mod tidy -diff +go mod verify +``` + +Expected: ordinary suite and all quality checks pass; module files are clean. + +### Phase 5 — full race gate and release + +Run from a reset PostgreSQL test database: + +```sh +make test-with-reset +make build +git diff --check +``` + +Expected: the complete serial race suite passes, all packages/examples build, +and the diff has no whitespace errors. Review the final diff specifically for +per-wait goroutines/connections, notification payload growth, application +payload logging, and any accidental schema change. + +After merge, tag `v0.5.0` (or the maintainer-approved next minor version) from +the reviewed commit. Record the exact tag and commit in Trails plan 012 before +that consumer upgrades. + +## 7. Test matrix + +| Case | Durable result | Wake path | Expected watch result | +|---|---|---|---| +| Event existed before `Watch` | event retained | none required | excluded by baseline | +| Event commits after `Watch` on same runtime | event retained | PostgreSQL hint | next typed record | +| Event commits on another runtime/pod | event retained | PostgreSQL broadcast | next typed record | +| Event commits while listener is disconnected | event retained | successful reconnect catch-up | next typed record | +| Listener outage exceeds caller bound | event retained | caller context | `ctx.Err()`; watch remains reusable and later reconnect finds the event | +| Notifications disabled | no watch created | validation | `ErrInvalid` | +| Separate writer has notifications disabled | event retained, no hint | caller context | no wake; unsupported deployment is characterized explicitly | +| Runtime not yet running | event may become durable | caller context, then listener startup catch-up | bounded `Next` may time out and remain reusable; after `Run`, next typed record | +| Event transaction rolls back | no event | no committed hint | continues waiting | +| Equivalent event redelivery | one event | no duplicate semantic hint | one record only | +| Different event name/run | other event retained | targeted/filtered | continues waiting | +| Run terminates | terminal projection/journal retained | terminal hint | `ErrTerminal` after matching-event drain | +| Terminal run is pruned after watch creation | run/history removed | terminal or reconnect hint | `ErrTerminal` | +| Live run is replaced | old terminal, new live holder | predecessor terminal hint | `ErrTerminal`; caller re-resolves | +| One `Next` context ends | no state change | context | `ctx.Err()`; watch remains until `Close` | +| Runtime stops | durable state unchanged | hub close | `ErrClosed` and cleanup | + +## 8. Done criteria + +- [x] `Event[T].Watch`, `EventWatch[T].Next`, and `Close` implement the exact + signatures and contract in Section 4.1; `EventWatch[T]` is the only new + exported type. +- [x] Future matching events come from durable journal reads, not notification + payloads or process memory. +- [x] One Flow runtime uses one existing dedicated listener connection for any + number of watches. +- [x] Cross-runtime tests prove a commit on A wakes a watch on B; no sticky + session assumption exists. +- [x] Application-event and terminal-only transactions emit a committed + run-scoped hint; runnable semantics are unchanged. +- [x] Listener startup/reconnect signals all registered watches and closes + commit-before-listen/disconnected windows through durable rereads. +- [x] Event watches reject notification-disabled runtimes; a pre-`Run` watch + becomes active through listener startup catch-up, with no event-watch + polling interval, fallback timer, or public fallback option. +- [x] Public documentation states that every writer of watched runs must have + notifications enabled; the disabled-writer characterization test proves + Flow does not silently promise a wake it cannot send. +- [x] An idle `Next` performs no database query after its initial read until an + explicit hint/startup/reconnect signal arrives. +- [x] Run replacement releases the old watch so callers can re-resolve a live + key. +- [x] No new command, wait row, lease, table, trigger, broker, or schema + migration is introduced. +- [x] `make test-with-reset`, `make build`, `go vet ./...`, formatting, module + consistency, and `git diff --check` all pass. +- [x] README and normative specs explain the read-side-only model and the + watch-before-read race closure. +- [ ] The reviewed release commit is tagged and available for Trails plan 012. + +## 9. STOP conditions + +Stop and report; do not improvise if: + +- correct decoding appears to require trusting a notification payload instead + of the durable journal, holding a database connection while waiting, or + creating durable state per watcher; +- any normal application-event or terminal commit can omit its transactional + notification wake when its writer has notifications enabled; +- the runtime cannot reject notifications-disabled/stopped watches while + retaining a bounded pre-`Run` watch for startup-race safety; +- application events can commit without an identifiable run ID; +- the journal query needs a schema/index change to stay bounded at the tested + 10,000-entry post-cursor shape; +- event/run hinting cannot preserve current immediate-runnable behavior with + the bounded per-semantic rules in Section 4.4 without moving a fault/commit + boundary or adding transaction-wide state; +- terminal replacement cannot wake the predecessor without changing live-key + or cancellation semantics; +- watcher cleanup requires one background goroutine per registered watch; +- the 10,000-entry sparse query is materially unbounded, or the 1,000-watcher + idle test shows timer-driven/query growth without an explicit signal; +- no-wait external event ingress regresses by more than 10% median in a + same-environment five-sample comparison and the regression cannot be + explained or corrected without expanding scope; +- a correct watch would require accepting a general `Client`, exposing its + private cursor/journal metadata, or adding another public option/type beyond + Section 4.1; or +- any test exposes a journal, fencing, lock-order, queue, or scheduler behavior + change outside this plan. + +## 10. Punchlist + +- [x] Characterize event-only, runnable-event, broadcast, and rollback hints. +- [x] Add typed cursor/next-event store reads and payload decoding. +- [x] Add only `EventWatch`, `Event.Watch`, `EventWatch.Next`, and `Close`; keep + cursor and journal metadata private. +- [x] Add the run-targeted local wake hub and listener routing. +- [x] Add the `event` notification kind with bounded per-semantic suppression; + rely on PostgreSQL for identical-payload folding. +- [x] Wake watches on every run terminal path and live-key replacement. +- [x] Prove cross-runtime, reconnect catch-up, cancellation, and shutdown + behavior under `-race`. +- [x] Document the API distinctions and watch-before-read recipe. +- [x] Measure sparse post-cursor reads, 1,000 idle watches, and event-ingress + overhead without adding speculative indexes or caches. +- [x] Run ordinary, race, build, vet, format, module, and diff gates. +- [ ] Tag the reviewed public feature release and record its commit for Trails. + +## 11. Maintenance notes + +- An event watch is an inspection primitive. If a future consumer asks Flow to + durably react, retry a callback, acknowledge offsets, select one of several + events, or keep a global subscription, that is a different architecture and + needs its own plan. +- Never recycle an event name for a changed payload schema. Watches decode with + the same definition contract as `WaitFor` and `Deliver`. +- Keep notification payloads identity-only. Applications may place sensitive + data in event bodies; it must remain in the durable database path and out of + `pg_notify`, logs, and observations. +- A deployment using watches keeps notifications enabled on every runtime that + writes the watched runs. Adding a poll-only writer later silently removes the + wake signal even though the event remains durable; review runtime options as + part of every deployment/configuration change. +- Event watches have no polling repair path. Consumers use bounded `Next` + contexts, re-read their projection after an event, and may read once when + their own response deadline expires. Listener startup/reconnect catch-up is + the only recovery wake for a commit made outside an active LISTEN session. +- Application event keys must identify one immutable fact. An empty payload + such as `None` cannot expose accidental key reuse, so an application whose + status can recur must include a stable causal identity or durable generation + in the key rather than relying only on the status text. +- Do not add a timer “just for safety.” If production evidence shows listener + health is insufficient, improve connection detection/observability or write + a new explicit plan; do not silently turn every waiter into a poller. diff --git a/specs/projects/flow/project_overview.md b/specs/projects/flow/project_overview.md index f7d2e12..54a9c92 100644 --- a/specs/projects/flow/project_overview.md +++ b/specs/projects/flow/project_overview.md @@ -142,8 +142,9 @@ Every semantic mutation is scoped to one run and locks its run row first. Within 2. allocates consecutive journal positions; 3. appends immutable semantic entries; 4. updates current-state, readiness, and delivery projections; -5. emits an optional transactional notification hint only when the mutation - creates immediately runnable work; and +5. emits a transactional run-identity hint when the mutation creates + immediately runnable work, records an application event, or terminalizes + the run and notifications are enabled; and 6. commits all changes together. The journal is gap-free and commit-ordered within each run. It records run start/failing, command creation, attempt start/conclusion, application events, and command/run terminal events. Current projections make claims and inspection efficient; replay verifies that retained semantic history reconstructs the same outcome. @@ -222,7 +223,9 @@ multi-migration development schema is not upgraded; operators drop and recreate the Flow schema first. `New` verifies schema compatibility and starts nothing. `Run` owns a bounded scheduler, lease renewal, wait/deadline/recovery maintenance, optional notification listening, observers, -and graceful shutdown. Polling is always sufficient for correctness. +and graceful shutdown. Polling remains sufficient for command processing; +EventWatch is a notification-backed inspection API with listener +startup/reconnect catch-up and no periodic polling. `PruneTerminalRuns` deletes one bounded batch of old terminal unkeyed or live-key aggregates in a Flow-owned transaction. Permanent non-empty keys and