From 7b9ea131a2d9f353d7cfbaf1f3ca242c60ccbde1 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 6 May 2026 21:10:33 -0400 Subject: [PATCH] feat(server/m7-scan-progress): stagePublisher helper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Persists a stage tally to scan_runs at a bounded cadence — every flushEvery items OR every flushPeriod elapsed, whichever fires first. Caller supplies marshal + persist closures so each stage owns its own tally type. Tick swallows mid-stage persist errors silently (observability bug, not scan failure); Flush propagates errors so the orchestrator's captureErr can surface them. Tests cover: count-based flush, time-based flush, Flush always persists, Tick swallows errors and keeps firing, Flush propagates errors, counters reset after each flush. --- internal/library/stage_publisher.go | 72 ++++++++++++ internal/library/stage_publisher_test.go | 134 +++++++++++++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 internal/library/stage_publisher.go create mode 100644 internal/library/stage_publisher_test.go diff --git a/internal/library/stage_publisher.go b/internal/library/stage_publisher.go new file mode 100644 index 00000000..ae80c8de --- /dev/null +++ b/internal/library/stage_publisher.go @@ -0,0 +1,72 @@ +package library + +import ( + "context" + "time" +) + +// stagePublisher persists a stage tally to scan_runs at a bounded +// cadence: every flushEvery items processed OR every flushPeriod +// elapsed, whichever fires first. The marshal closure produces the +// current tally JSON; persist writes it. +// +// Usage pattern: +// +// pub := newStagePublisher(ctx, 500, time.Second, marshalFn, persistFn) +// defer func() { +// if err := pub.Flush(); err != nil { /* route to captureErr */ } +// }() +// for ... { +// processOne() +// pub.Tick() +// } +type stagePublisher struct { + ctx context.Context + flushEvery int + flushPeriod time.Duration + marshal func() []byte + persist func([]byte) error + sinceItem int + lastFlush time.Time +} + +// newStagePublisher constructs a publisher. The ctx is captured for +// future use (currently unused in the helper itself; persist closures +// typically capture their own context via the caller). +func newStagePublisher(ctx context.Context, every int, period time.Duration, + marshal func() []byte, persist func([]byte) error) *stagePublisher { + return &stagePublisher{ + ctx: ctx, + flushEvery: every, + flushPeriod: period, + marshal: marshal, + persist: persist, + lastFlush: time.Now(), + } +} + +// Tick records one item processed. If the flushEvery count is reached +// or flushPeriod has elapsed since the last flush, it triggers a +// persist. Mid-stage flush errors are silently suppressed — partial- +// tally write failures are observability bugs, not scan failures, and +// the next Tick retries. +func (p *stagePublisher) Tick() { + p.sinceItem++ + if p.sinceItem >= p.flushEvery || time.Since(p.lastFlush) >= p.flushPeriod { + _ = p.flush() + } +} + +// Flush writes the current tally unconditionally and returns any +// persist error. Call at stage end via a deferred wrapper that +// routes the error to the orchestrator's captureErr — preserves +// today's behavior where final-write failures are diagnostic. +func (p *stagePublisher) Flush() error { return p.flush() } + +func (p *stagePublisher) flush() error { + blob := p.marshal() + err := p.persist(blob) + p.sinceItem = 0 + p.lastFlush = time.Now() + return err +} diff --git a/internal/library/stage_publisher_test.go b/internal/library/stage_publisher_test.go new file mode 100644 index 00000000..ffbe969d --- /dev/null +++ b/internal/library/stage_publisher_test.go @@ -0,0 +1,134 @@ +package library + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" +) + +// Helper: build a publisher that records persist calls into atomic counters. +func newTestPublisher(t *testing.T, flushEvery int, flushPeriod time.Duration) (*stagePublisher, *atomic.Int32, *atomic.Int32) { + t.Helper() + var marshalCalls, persistCalls atomic.Int32 + pub := newStagePublisher(context.Background(), flushEvery, flushPeriod, + func() []byte { + marshalCalls.Add(1) + return []byte(`{}`) + }, + func(_ []byte) error { + persistCalls.Add(1) + return nil + }, + ) + return pub, &marshalCalls, &persistCalls +} + +func TestStagePublisher_Tick_FlushesAtCountThreshold(t *testing.T) { + pub, _, persists := newTestPublisher(t, 3, 1*time.Hour) // huge time so only count fires + + for i := 0; i < 3; i++ { + pub.Tick() + } + if persists.Load() != 1 { + t.Errorf("persists after 3 ticks = %d, want 1", persists.Load()) + } + + for i := 0; i < 3; i++ { + pub.Tick() + } + if persists.Load() != 2 { + t.Errorf("persists after 6 ticks = %d, want 2", persists.Load()) + } +} + +func TestStagePublisher_Tick_FlushesOnTimeout(t *testing.T) { + pub, _, persists := newTestPublisher(t, 10000, 10*time.Millisecond) // huge count so only time fires + + pub.Tick() + if persists.Load() != 0 { + t.Errorf("first tick should not flush; persists = %d", persists.Load()) + } + + time.Sleep(50 * time.Millisecond) + pub.Tick() // time-based flush should fire here + if persists.Load() < 1 { + t.Errorf("persists after timeout tick = %d, want >= 1", persists.Load()) + } +} + +func TestStagePublisher_Flush_AlwaysPersists(t *testing.T) { + pub, _, persists := newTestPublisher(t, 1000, 1*time.Hour) + + pub.Tick() // doesn't flush — count too low, time too long + + if err := pub.Flush(); err != nil { + t.Errorf("Flush() err = %v, want nil", err) + } + if persists.Load() != 1 { + t.Errorf("persists after Flush = %d, want 1", persists.Load()) + } +} + +func TestStagePublisher_Tick_SwallowsPersistErrors(t *testing.T) { + persistCalls := atomic.Int32{} + pub := newStagePublisher(context.Background(), 1, 1*time.Hour, + func() []byte { return []byte(`{}`) }, + func(_ []byte) error { + persistCalls.Add(1) + return errors.New("simulated DB failure") + }, + ) + + // Ticks should not panic, should keep firing the persist (which keeps erroring). + for i := 0; i < 5; i++ { + pub.Tick() + } + if persistCalls.Load() != 5 { + t.Errorf("persistCalls after 5 ticks = %d, want 5 (errors should not stop subsequent ticks)", persistCalls.Load()) + } +} + +func TestStagePublisher_Flush_PropagatesPersistError(t *testing.T) { + wantErr := errors.New("DB exploded") + pub := newStagePublisher(context.Background(), 1, 1*time.Hour, + func() []byte { return []byte(`{}`) }, + func(_ []byte) error { return wantErr }, + ) + + if err := pub.Flush(); !errors.Is(err, wantErr) { + t.Errorf("Flush() err = %v, want %v", err, wantErr) + } +} + +func TestStagePublisher_FlushResetsCounters(t *testing.T) { + pub, _, persists := newTestPublisher(t, 5, 1*time.Hour) + + for i := 0; i < 4; i++ { + pub.Tick() + } + if persists.Load() != 0 { + t.Errorf("persists at 4 ticks = %d, want 0", persists.Load()) + } + + if err := pub.Flush(); err != nil { + t.Fatal(err) + } + if persists.Load() != 1 { + t.Fatalf("persists after Flush = %d, want 1", persists.Load()) + } + + // After Flush, the counter resets — need 5 more ticks to flush again. + for i := 0; i < 4; i++ { + pub.Tick() + } + if persists.Load() != 1 { + t.Errorf("persists after Flush + 4 ticks = %d, want 1 (counter should have reset)", persists.Load()) + } + + pub.Tick() // 5th tick after reset → flush + if persists.Load() != 2 { + t.Errorf("persists after Flush + 5 ticks = %d, want 2", persists.Load()) + } +}