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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W1rkJ2hBMybnJctPbcy4tT
This commit is contained in:
kami
2026-07-27 23:19:15 +04:00
parent 972845bd98
commit 1f46a34afb
5 changed files with 150 additions and 5 deletions
+18 -1
View File
@@ -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 {
+61
View File
@@ -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)
}
}