Add distributed application event watches

.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 ./...

README.md

@@ -202,6 +202,50 @@ 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. 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.

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 {

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)
 	}

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()
+}