From 1f46a34afb2095ad0251e83fbd5cd1694c46fe5e Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 27 Jul 2026 23:19:15 +0400 Subject: [PATCH] fix(delivery): stop killing the fanout goroutine on a single send error Closes S4 (AUDIT.md): Fanout.Run returned on the first sender error, permanently ending notifications for the process lifetime after one ntfy hiccup. Failed sends now go through an OnError hook and the loop continues. Also persists the delivery cursor to a file next to ORCHESTRA_DATA so a restart resumes from the last delivered event instead of re-notifying the entire log from seq 0. Adds the package's first test. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01W1rkJ2hBMybnJctPbcy4tT --- AUDIT.md | 16 ++++++++ cmd/orchestra/main.go | 18 ++++++++- internal/delivery/delivery.go | 19 +++++++++- internal/delivery/delivery_test.go | 61 ++++++++++++++++++++++++++++++ progress.md | 41 ++++++++++++++++++-- 5 files changed, 150 insertions(+), 5 deletions(-) create mode 100644 internal/delivery/delivery_test.go diff --git a/AUDIT.md b/AUDIT.md index 3b02c21..815145a 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -824,6 +824,22 @@ which emits the same events live. it requires intercepting a response — an opencode plugin (`~/.config/opencode/plugin/`) is the only clean hook. +### S4 — closed, 2026-07-27 + +`delivery.Fanout.Run` no longer `return`s on the first sender error — a +single ntfy hiccup used to permanently kill the notification goroutine for +the rest of the process (`main.go` only logged the `Run` error, it never +restarted the goroutine). Failed sends now go through an `OnError` hook +(default `log.Printf`), and the loop keeps going to the next sender/event. +Cursor persistence was also added: `SaveCursor` is called every time the +cursor advances, and `main.go` wires it to a `delivery-cursor` file next to +`ORCHESTRA_DATA`, loaded on startup — a restart resumes from the last +delivered event instead of re-notifying the entire log from seq 0. Covered +by `TestFanoutContinuesAfterSendError` (`internal/delivery/delivery_test.go` +— previously the package had zero tests): a failing sender and a healthy +sender both receive the event, the cursor still advances, and `Run` only +exits on context cancellation, never on the send error. + ### Design consequences (not yet implemented) 1. **Percentages are a level, not a delta.** diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 912ee75..e666917 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -852,8 +852,24 @@ func main() { senders = append(senders, delivery.Ntfy{Topic: topic, Token: os.Getenv("ORCHESTRA_NTFY_TOKEN"), URL: os.Getenv("ORCHESTRA_NTFY_URL")}) } if len(senders) > 0 { + cursorPath := filepath.Join(dir, "delivery-cursor") + var startCursor uint64 + if b, err := os.ReadFile(cursorPath); err == nil { + if v, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64); err == nil { + startCursor = v + } + } go func() { - err := (&delivery.Fanout{Senders: senders}).Run(context.Background(), s.Events) + fanout := &delivery.Fanout{ + Senders: senders, + Cursor: startCursor, + SaveCursor: func(cursor uint64) { + if err := os.WriteFile(cursorPath, []byte(strconv.FormatUint(cursor, 10)), 0o644); err != nil { + log.Printf("delivery cursor persist: %v", err) + } + }, + } + err := fanout.Run(context.Background(), s.Events) if err != nil { log.Printf("delivery fanout: %v", err) } diff --git a/internal/delivery/delivery.go b/internal/delivery/delivery.go index 05afce5..6d17c6d 100644 --- a/internal/delivery/delivery.go +++ b/internal/delivery/delivery.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "log" "net/http" "orchestra/internal/domain" "time" @@ -101,23 +102,39 @@ type Fanout struct { Senders []Sender Cursor uint64 Retry time.Duration + // OnError is called for each failed send instead of aborting the fanout + // goroutine (S4, AUDIT.md: a single ntfy hiccup used to `return` and + // permanently kill notifications for the rest of the process). Defaults + // to log.Printf. + OnError func(err error) + // SaveCursor, if set, is called after the cursor advances past a + // processed event so a restart can resume from here instead of + // re-notifying the entire log from seq 0 (S4). + SaveCursor func(cursor uint64) } func (f *Fanout) Run(ctx context.Context, events func(uint64) []domain.Event) error { if f.Retry <= 0 { f.Retry = 5 * time.Second } + onError := f.OnError + if onError == nil { + onError = func(err error) { log.Printf("delivery: %v", err) } + } for { for _, e := range events(f.Cursor) { if msg, ok := Message(e); ok { for _, s := range f.Senders { if err := s.Send(ctx, msg); err != nil { - return err + onError(fmt.Errorf("send %s to %T: %w", msg, s, err)) } } } if e.Seq > f.Cursor { f.Cursor = e.Seq + if f.SaveCursor != nil { + f.SaveCursor(f.Cursor) + } } } select { diff --git a/internal/delivery/delivery_test.go b/internal/delivery/delivery_test.go new file mode 100644 index 0000000..4dd356e --- /dev/null +++ b/internal/delivery/delivery_test.go @@ -0,0 +1,61 @@ +package delivery_test + +import ( + "context" + "errors" + "orchestra/internal/delivery" + "orchestra/internal/domain" + "testing" + "time" +) + +type fakeSender struct { + fail bool + sent []string +} + +func (f *fakeSender) Send(_ context.Context, msg string) error { + if f.fail { + return errors.New("boom") + } + f.sent = append(f.sent, msg) + return nil +} + +// TestFanoutContinuesAfterSendError guards S4: a single sender error must +// not kill the fanout goroutine or stall the cursor for the other senders. +func TestFanoutContinuesAfterSendError(t *testing.T) { + failing := &fakeSender{fail: true} + ok := &fakeSender{} + var gotErrs int + var savedCursor uint64 + f := &delivery.Fanout{ + Senders: []delivery.Sender{failing, ok}, + Retry: time.Millisecond, + OnError: func(error) { gotErrs++ }, + SaveCursor: func(c uint64) { savedCursor = c }, + } + events := []domain.Event{{Seq: 1, Type: "TaskCompleted", TaskID: "t1"}} + served := false + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := f.Run(ctx, func(cursor uint64) []domain.Event { + if served { + return nil + } + served = true + return events + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Run returned %v, want deadline exceeded (goroutine must survive send errors)", err) + } + if gotErrs == 0 { + t.Fatal("expected OnError to be called for the failing sender") + } + if len(ok.sent) != 1 { + t.Fatalf("healthy sender got %d messages, want 1", len(ok.sent)) + } + if savedCursor != 1 { + t.Fatalf("cursor=%d, want 1 (must advance despite the other sender's error)", savedCursor) + } +} diff --git a/progress.md b/progress.md index 3778e75..7c28970 100644 --- a/progress.md +++ b/progress.md @@ -172,9 +172,44 @@ Fixed so far: in-pane prompt to write the file before exiting, which is a live-deployment fact, not something provable from source. -Not yet started: B7 (quota projection has no producer), -Codex/opencode completion producers, the turn-decision endpoint, S2–S4, -S7–S11. See `AUDIT.md` for the full plan. +- **B7 (post-hoc producer) + Phase 2 turn-decision endpoint** — landed + together, since both are new `QuotaReported`/turn-boundary paths off the + same completion/turn events. `POST /v1/harness/complete` now appends a + `QuotaReported` event (`harness_id` from the closing lease, `consumed` + from the same `usage.Numerator()` used for the receipt), so the router's + 5h/weekly availability filter and the brief's `quota_consumed` stop + evaluating against a permanent zero. New `Coordinator.TurnDecision` + (`internal/orchestrator/orchestrator.go`) mirrors `rotate()`'s per-task + logic (occupancy → turn-boundary → handoff-file → release) but runs + synchronously once per turn instead of waiting for `Monitor`'s ticker, + returning one of `continue`/`prepare_handoff`/`rotate_now`/`refuse` via the + new `POST /v1/harness/turn`. The Claude Stop hook + (`deploy/hooks/orchestra-stop.sh`) now calls this endpoint on every + ordinary turn boundary (report marker absent) instead of no-op'ing, and + exits 2 on `refuse` to stop the harness from finishing an unsafe turn. + Covered by `TestTurnDecision` (`internal/orchestrator/rotation_test.go`): + continue-below-threshold, refuse-when-not-at-boundary, and + rotate_now-releases-and-emits-a-valid-TaskReleased cases. + **Not done:** live per-harness *push* producers (Claude statusline, + Codex rollout tail) that would give B7 a second, continuous producer + independent of task completion — recorded as a design investigation in + AUDIT.md ("Real harness quota sources") but not implemented; Codex/ + opencode's own equivalents of the Claude Stop hook (whether their + turn-boundary mechanism actually calls `/v1/harness/turn`) also remain + unbuilt, same caveat as Phase 2 item 4 already named for `/complete`. + +- **S4** — `delivery.Fanout.Run` used to `return` on the first sender error, + permanently killing the notification goroutine (a single ntfy hiccup meant + no notifications for the rest of the process's lifetime, since nothing + restarts it). Failed sends now go through an `OnError` hook instead of + aborting the loop. Cursor is also persisted now (`SaveCursor` → a + `delivery-cursor` file next to `ORCHESTRA_DATA`, loaded on startup), so a + restart resumes from the last delivered event instead of re-notifying the + entire log from seq 0. `internal/delivery` previously had zero tests; + added `TestFanoutContinuesAfterSendError`. + +Not yet started: Codex/opencode completion producers, S2–S3, S7–S11. See +`AUDIT.md` for the full plan. **Phase 0 done (2026-07-27):** this box has live TCP reachability to the real herdr instance at `192.168.1.105:9245` — verified by hand (raw JSON-RPC