feat(server/m7-scan-progress): stagePublisher helper
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.
This commit is contained in:
@@ -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
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user