amadan

carlos / platform Public

Retry final-hour storage collection and retain failed publications

claude 1dea9cafa072d3ff893c6745c0895ea48b8fce01 parent 988a052
3 files changed, +617 −15
  • internal/console/economicspass.go +53 −15
  • internal/console/economicstorageretry.go +126 −0
  • internal/console/economicstorageretry_test.go +438 −0

CI passed — run details

diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go
index ffdb42e62..80076fb06 100644
--- a/internal/console/economicspass.go
+++ b/internal/console/economicspass.go
@@ -33,15 +33,25 @@ import (
// the day is written — and catches up promptly after a console restart.
const economicsPassInterval = time.Hour
-// StartEconomicsPass runs economicsPass immediately and then hourly until
-// ctx is done. Started from cmd/console's serve(), deliberately NOT from
+// StartEconomicsPass runs the hourly stages and an independent storage
+// retry worker until ctx is done. Started from cmd/console's serve(), NOT from
// NewServer — the StartAlertWatcher rule: test servers never grow
// goroutines (alertwatch.go:44).
func (s *Server) StartEconomicsPass(ctx context.Context) {
+ ctx, cancel := context.WithCancel(ctx)
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ s.storageRetryRunner().run(ctx)
+ }()
+ defer func() {
+ cancel()
+ <-done
+ }()
t := time.NewTicker(economicsPassInterval)
defer t.Stop()
for {
- if err := s.economicsPass(ctx, time.Now()); err != nil {
+ if err := s.runEconomicsPass(ctx, time.Now(), false); err != nil {
log.Printf("console: economics pass: %v", err)
}
select {
@@ -57,16 +67,34 @@ func (s *Server) StartEconomicsPass(ctx context.Context) {
// a failed stage logs and the next still runs, because a missing census day
// must not cost the month's statements (spec: error handling).
func (s *Server) economicsPass(ctx context.Context, now time.Time) error {
+ return s.runEconomicsPass(ctx, now, true)
+}
+
+func (s *Server) runEconomicsPass(ctx context.Context, now time.Time, storage bool) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
date := usage.DayOf(now)
if err := s.censusPass(ctx, date); err != nil {
log.Printf("console: census pass %s: %v", date, err)
}
- if err := s.storagePass(ctx, date); err != nil {
- log.Printf("console: storage pass %s: %v", date, err)
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if storage {
+ if err := s.storagePass(ctx, date); err != nil {
+ log.Printf("console: storage pass %s: %v", date, err)
+ }
+ }
+ if err := ctx.Err(); err != nil {
+ return err
}
if err := s.statementsPass(ctx, now); err != nil {
log.Printf("console: statements pass: %v", err)
}
+ if err := ctx.Err(); err != nil {
+ return err
+ }
if err := s.billPass(ctx, now); err != nil {
log.Printf("console: bill pass: %v", err)
}
@@ -165,14 +193,25 @@ func (s *Server) censusPass(ctx context.Context, date string) error {
// bucket to support sized listings (release.SizeLister); a store that
// cannot answer marks the day's sweep absent rather than lying with zeros.
func (s *Server) storagePass(ctx context.Context, date string) error {
+ day, err := s.collectStorageDay(ctx, date)
+ if err != nil || day == nil {
+ return err
+ }
+ _, err = usage.CreateDayObject(ctx, s.cfg.Releases, usage.StorageDayKey(date), day)
+ return err
+}
+
+func (s *Server) collectStorageDay(ctx context.Context, date string) (*usage.StorageDay, error) {
key := usage.StorageDayKey(date)
if _, _, err := s.cfg.Releases.Get(ctx, key); err == nil {
- return nil // today already written (by us or a sibling console)
+ return nil, nil // today already written (by us or a sibling console)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return nil, err
}
sizer, ok := s.cfg.Releases.(release.SizeLister)
if !ok {
log.Printf("console: storage pass: bucket store does not support sized listings; skipping")
- return nil
+ return nil, nil
}
// ListDirs (both DirStore and S3Store) returns each child as a FULL,
// slash-terminated prefix — "apps/bab/", not "bab" — so the next level's
@@ -180,7 +219,7 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
// name for map keys/AppPrefix is recovered by trimming both ends.
acctPrefixes, err := s.cfg.Releases.ListDirs(ctx, "apps/")
if err != nil {
- return err
+ return nil, err
}
// Who holds which bucket, read once for the whole pass, with each
// bucket's ledger generation at that moment.
@@ -203,7 +242,7 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
})
if err != nil {
if !storageLastChance(date) {
- return fmt.Errorf("storage pass: %w", err)
+ return nil, fmt.Errorf("storage pass: %w", err)
}
log.Printf("console: storage pass: %v; the day is ending, so it is written with every store held", err)
inventory = nil
@@ -220,13 +259,13 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
acct := strings.TrimSuffix(strings.TrimPrefix(acctPrefix, "apps/"), "/")
appPrefixes, err := s.cfg.Releases.ListDirs(ctx, acctPrefix)
if err != nil {
- return err
+ return nil, err
}
for _, appPrefix := range appPrefixes {
app := strings.TrimSuffix(strings.TrimPrefix(appPrefix, acctPrefix), "/")
sizes, err := sizer.ListSizes(ctx, release.AppPrefix(acct, app))
if err != nil {
- return err
+ return nil, err
}
var st usage.AppStorage
for _, ks := range sizes {
@@ -247,7 +286,7 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
// A reading the store failed to return, with time left in
// the day: the whole pass is tried again, and no store is
// held on it.
- return fmt.Errorf("storage pass: store record %s/%s: %w", acct, app, gerr)
+ return nil, fmt.Errorf("storage pass: store record %s/%s: %w", acct, app, gerr)
}
if gerr != nil {
// Absent is (nil, "", nil) — objstore.Get's own idiom — so a
@@ -284,7 +323,7 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
if holdEveryStore {
st.StoreUnswept = true
} else if held, herr := s.sizeSoleHolder(ctx, inventory, rec, acct, app, bucket, date, &st); herr != nil {
- return herr
+ return nil, herr
} else if held != "" {
log.Printf("console: storage pass: %s/%s: %s; held", acct, app, held)
st.StoreUnswept = true
@@ -293,8 +332,7 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
day.Apps[acct+"/"+app] = st
}
}
- _, err = usage.CreateDayObject(ctx, s.cfg.Releases, key, day)
- return err
+ return &day, nil
}
// dayCensus is one UTC day's census in the shape sweepUsage's consumers
diff --git a/internal/console/economicstorageretry.go b/internal/console/economicstorageretry.go
new file mode 100644
index 000000000..7bcb7b05d
--- /dev/null
+++ b/internal/console/economicstorageretry.go
@@ -0,0 +1,126 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+const (
+ storageRetryInterval = time.Minute
+ storageOperationTimeout = 2 * time.Minute
+ storagePublicationGrace = 5 * time.Minute
+)
+
+type storageRetryRunner struct {
+ now func() time.Time
+ wait func(context.Context, time.Duration) bool
+ limit func(context.Context, time.Duration) (context.Context, context.CancelFunc)
+ collect func(context.Context, string) (*usage.StorageDay, error)
+ publish func(context.Context, string, []byte) error
+}
+
+func (s *Server) storageRetryRunner() storageRetryRunner {
+ return storageRetryRunner{
+ now: storageClock,
+ wait: func(ctx context.Context, d time.Duration) bool {
+ timer := time.NewTimer(d)
+ defer timer.Stop()
+ select {
+ case <-ctx.Done():
+ return false
+ case <-timer.C:
+ return true
+ }
+ },
+ limit: context.WithTimeout,
+ collect: s.collectStorageDay,
+ publish: func(ctx context.Context, date string, body []byte) error {
+ _, err := s.cfg.Releases.Put(ctx, usage.StorageDayKey(date), body, "")
+ if errors.Is(err, release.ErrCASConflict) {
+ return nil
+ }
+ return err
+ },
+ }
+}
+
+func (r storageRetryRunner) run(ctx context.Context) {
+ var body []byte
+ var date, finished string
+ var expires time.Time
+ for ctx.Err() == nil {
+ now := r.now().UTC()
+ today := usage.DayOf(now)
+ end := now.Truncate(24 * time.Hour).Add(24 * time.Hour)
+ if body != nil && !now.Before(expires) {
+ body = nil
+ }
+ if body == nil {
+ if today == finished {
+ if !r.wait(ctx, end.Sub(now)) {
+ return
+ }
+ continue
+ }
+ date = today
+ opCtx, cancel := r.limit(ctx, min(storageOperationTimeout, end.Sub(now)))
+ day, err := r.collect(opCtx, date)
+ if err == nil {
+ err = opCtx.Err()
+ }
+ cancel()
+ collected := r.now().UTC()
+ if err == nil && (usage.DayOf(collected) != date || !collected.Before(end)) {
+ err = context.DeadlineExceeded
+ }
+ if err != nil {
+ log.Printf("console: storage pass %s: %v", date, err)
+ } else if day == nil {
+ finished = date
+ continue
+ } else {
+ body, err = json.Marshal(day)
+ if err != nil {
+ log.Printf("console: storage pass %s: %v", date, err)
+ }
+ expires = end.Add(storagePublicationGrace)
+ }
+ }
+ if body != nil && ctx.Err() == nil {
+ now = r.now().UTC()
+ if now.Before(expires) {
+ opCtx, cancel := r.limit(ctx, min(storageOperationTimeout, expires.Sub(now)))
+ err := r.publish(opCtx, date, body)
+ cancel()
+ if err == nil {
+ body = nil
+ finished = date
+ continue
+ }
+ log.Printf("console: storage pass %s: %v", date, err)
+ }
+ }
+ now = r.now().UTC()
+ end = now.Truncate(24 * time.Hour).Add(24 * time.Hour)
+ delay := economicsPassInterval
+ finalHour := end.Add(-time.Hour)
+ if !now.Before(finalHour) || body != nil && date != usage.DayOf(now) {
+ delay = storageRetryInterval
+ } else if finalHour.Sub(now) < delay {
+ delay = finalHour.Sub(now)
+ }
+ delay = min(delay, end.Sub(now))
+ if body != nil {
+ delay = min(delay, max(0, expires.Sub(now)))
+ }
+ if !r.wait(ctx, delay) {
+ return
+ }
+ }
+}
diff --git a/internal/console/economicstorageretry_test.go b/internal/console/economicstorageretry_test.go
new file mode 100644
index 000000000..30e13cd2a
--- /dev/null
+++ b/internal/console/economicstorageretry_test.go
@@ -0,0 +1,438 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "io"
+ "os"
+ "reflect"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+type storageRetryTestStore struct {
+ release.Store
+ get func(context.Context, string) ([]byte, string, error)
+ put func(context.Context, string, []byte, string) (string, error)
+ dirs func(context.Context, string) ([]string, error)
+ sizes func(context.Context, string) ([]release.KeySize, error)
+}
+
+func (s *storageRetryTestStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if s.get != nil {
+ return s.get(ctx, key)
+ }
+ return s.Store.Get(ctx, key)
+}
+func (s *storageRetryTestStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ if s.put != nil {
+ return s.put(ctx, key, body, etag)
+ }
+ return s.Store.Put(ctx, key, body, etag)
+}
+func (s *storageRetryTestStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
+ if s.dirs != nil {
+ return s.dirs(ctx, prefix)
+ }
+ return s.Store.ListDirs(ctx, prefix)
+}
+func (s *storageRetryTestStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
+ if s.sizes != nil {
+ return s.sizes(ctx, prefix)
+ }
+ return s.Store.(release.SizeLister).ListSizes(ctx, prefix)
+}
+
+func storageRetryFixture(t *testing.T, now *time.Time) (*consoleFixture, *storageRetryTestStore, storageRetryRunner) {
+ t.Helper()
+ fx := newConsoleFixture(t)
+ fx.srv.ceCosts = func(context.Context, time.Time, time.Time) (map[string]float64, error) {
+ return nil, io.ErrUnexpectedEOF
+ }
+ store := &storageRetryTestStore{Store: fx.rel}
+ fx.srv.cfg.Releases = store
+ previous := storageClock
+ storageClock = func() time.Time { return *now }
+ t.Cleanup(func() { storageClock = previous })
+ runner := fx.srv.storageRetryRunner()
+ return fx, store, runner
+}
+
+func TestStorageRetryWakesAtFinalHour(t *testing.T) {
+ now := time.Date(2026, 10, 3, 22, 30, 0, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ failures := 3
+ store.dirs = func(ctx context.Context, prefix string) ([]string, error) {
+ if prefix == "apps/" && failures > 0 {
+ failures--
+ return nil, io.ErrUnexpectedEOF
+ }
+ return store.Store.ListDirs(ctx, prefix)
+ }
+ var waits []time.Duration
+ runner.wait = func(ctx context.Context, d time.Duration) bool {
+ if _, _, err := fx.rel.Get(ctx, usage.StorageDayKey("2026-10-03")); err == nil {
+ return false
+ }
+ waits = append(waits, d)
+ if len(waits) > 4 {
+ t.Fatal("retry did not publish")
+ }
+ now = now.Add(d)
+ return true
+ }
+ runner.run(t.Context())
+ if !reflect.DeepEqual(waits, []time.Duration{30 * time.Minute, time.Minute, time.Minute}) {
+ t.Fatalf("wake delays %v", waits)
+ }
+ if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03")); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestStorageRetryRetainsBytesAcrossMidnight(t *testing.T) {
+ for _, lostResponse := range []bool{false, true} {
+ t.Run(map[bool]string{false: "failed write", true: "lost answer then CAS"}[lostResponse], func(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 59, 30, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ var bodies [][]byte
+ var keys []string
+ listings := 0
+ atFirstPut := 0
+ store.sizes = func(ctx context.Context, prefix string) ([]release.KeySize, error) {
+ listings++
+ return store.Store.(release.SizeLister).ListSizes(ctx, prefix)
+ }
+ store.put = func(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ keys = append(keys, key)
+ bodies = append(bodies, append([]byte(nil), body...))
+ if etag != "" {
+ t.Fatal("publication was not create-only")
+ }
+ if len(bodies) == 1 {
+ atFirstPut = listings
+ if lostResponse {
+ if _, err := store.Store.Put(ctx, key, body, etag); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if _, err := fx.rel.Put(ctx, "apps/"+fixtureAcct+"/"+fixtureApp+"/late", make([]byte, 1000), ""); err != nil {
+ t.Fatal(err)
+ }
+ return "", io.ErrUnexpectedEOF
+ }
+ out, err := store.Store.Put(ctx, key, body, etag)
+ return out, err
+ }
+ runner.wait = func(ctx context.Context, d time.Duration) bool {
+ if len(bodies) == 2 {
+ cancel()
+ return false
+ }
+ if d != 30*time.Second {
+ t.Fatalf("delay at midnight %v", d)
+ }
+ now = now.Add(d)
+ return true
+ }
+ // Stop before a fresh current-day collection, after the old day's successful publication.
+ collect := runner.collect
+ runner.collect = func(ctx context.Context, date string) (*usage.StorageDay, error) {
+ if date != "2026-10-03" {
+ cancel()
+ return nil, context.Canceled
+ }
+ return collect(ctx, date)
+ }
+ runner.run(ctx)
+ if len(bodies) != 2 || string(bodies[0]) != string(bodies[1]) {
+ t.Fatal("publication retry changed the collected payload")
+ }
+ if listings != atFirstPut || listings == 0 {
+ t.Fatalf("old date recollected: first=%d final=%d", atFirstPut, listings)
+ }
+ if keys[0] != usage.StorageDayKey("2026-10-03") || keys[1] != keys[0] {
+ t.Fatalf("publication dates %v", keys)
+ }
+ raw, _, err := fx.rel.Get(t.Context(), keys[0])
+ if err != nil || string(raw) != string(bodies[0]) {
+ t.Fatalf("published payload changed: %v", err)
+ }
+ })
+ }
+}
+
+func TestStorageRetryGraceExpiresWithoutYesterdayMeasurements(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 59, 30, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ var collected []string
+ collect := runner.collect
+ runner.collect = func(ctx context.Context, date string) (*usage.StorageDay, error) {
+ collected = append(collected, date)
+ return collect(ctx, date)
+ }
+ oldWrites := 0
+ store.put = func(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ if key == usage.StorageDayKey("2026-10-03") {
+ oldWrites++
+ if !now.Before(time.Date(2026, 10, 4, 0, 5, 0, 0, time.UTC)) {
+ t.Fatal("old payload retried past grace")
+ }
+ return "", io.ErrUnexpectedEOF
+ }
+ return store.Store.Put(ctx, key, body, etag)
+ }
+ waits := 0
+ runner.wait = func(ctx context.Context, d time.Duration) bool {
+ if _, _, err := fx.rel.Get(ctx, usage.StorageDayKey("2026-10-04")); err == nil {
+ return false
+ }
+ waits++
+ if waits > 8 {
+ t.Fatal("grace did not expire")
+ }
+ now = now.Add(d)
+ return true
+ }
+ runner.run(t.Context())
+ if oldWrites != 6 || !reflect.DeepEqual(collected, []string{"2026-10-03", "2026-10-04"}) {
+ t.Fatalf("writes=%d collections=%v", oldWrites, collected)
+ }
+ if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("fabricated yesterday: %v", err)
+ }
+}
+
+func TestStorageRetryDiscardsCollectionFinishingAfterMidnight(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 59, 30, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ limits := []time.Duration{}
+ runner.limit = func(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+ limits = append(limits, d)
+ return context.WithCancel(ctx)
+ }
+ jumped := false
+ store.sizes = func(ctx context.Context, prefix string) ([]release.KeySize, error) {
+ if !jumped {
+ jumped = true
+ now = now.Add(time.Minute)
+ }
+ return store.Store.(release.SizeLister).ListSizes(ctx, prefix)
+ }
+ waits := 0
+ runner.wait = func(ctx context.Context, d time.Duration) bool {
+ waits++
+ if waits > 2 {
+ t.Fatal("did not finish current day")
+ }
+ if _, _, err := fx.rel.Get(ctx, usage.StorageDayKey("2026-10-04")); err == nil {
+ return false
+ }
+ now = now.Add(d)
+ return true
+ }
+ runner.run(t.Context())
+ if len(limits) < 1 || limits[0] != 30*time.Second {
+ t.Fatalf("collection was not bounded by midnight: %v", limits)
+ }
+ if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("late collection labeled yesterday: %v", err)
+ }
+ if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-04")); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestStorageRetryConcurrentCreatorWins(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 30, 0, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ winner := []byte(`{"v":1,"date":"2026-10-03","apps":{}}`)
+ puts := 0
+ store.put = func(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ puts++
+ if _, err := store.Store.Put(ctx, key, winner, ""); err != nil {
+ t.Fatal(err)
+ }
+ return store.Store.Put(ctx, key, body, etag)
+ }
+ runner.wait = func(_ context.Context, d time.Duration) bool {
+ if d != 30*time.Minute {
+ t.Fatalf("CAS loser scheduled retry: %v", d)
+ }
+ return false
+ }
+ runner.run(t.Context())
+ raw, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03"))
+ if puts != 1 || err != nil || string(raw) != string(winner) {
+ t.Fatalf("concurrent winner overwritten: puts=%d err=%v", puts, err)
+ }
+}
+
+func TestStorageRetryCollectionCancellation(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 30, 0, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ entered := make(chan struct{})
+ opCancel := make(chan context.CancelFunc, 1)
+ var once sync.Once
+ store.dirs = func(ctx context.Context, prefix string) ([]string, error) {
+ blocked := false
+ once.Do(func() { blocked = true; close(entered) })
+ if blocked {
+ <-ctx.Done()
+ return nil, ctx.Err()
+ }
+ return store.Store.ListDirs(ctx, prefix)
+ }
+ runner.limit = func(parent context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+ if d != 2*time.Minute {
+ t.Errorf("unbounded operation: %v", d)
+ }
+ op, stop := context.WithCancel(parent)
+ select {
+ case opCancel <- stop:
+ default:
+ }
+ return op, stop
+ }
+ runner.wait = func(ctx context.Context, d time.Duration) bool {
+ if _, _, err := fx.rel.Get(ctx, usage.StorageDayKey("2026-10-03")); err == nil {
+ return false
+ }
+ now = now.Add(d)
+ return true
+ }
+ done := make(chan struct{})
+ go func() { defer close(done); runner.run(ctx) }()
+ <-entered
+ (<-opCancel)()
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("canceled collection kept retry owner blocked")
+ }
+ if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03")); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestStorageRetryIndependentOfHungCensus(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.ceCosts = func(context.Context, time.Time, time.Time) (map[string]float64, error) {
+ return nil, io.ErrUnexpectedEOF
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ census := make(chan struct{})
+ published := make(chan struct{})
+ var censusOnce, publishOnce sync.Once
+ store := &storageRetryTestStore{Store: fx.rel}
+ store.get = func(ctx context.Context, key string) ([]byte, string, error) {
+ if key == usage.CensusDayKey(usage.DayOf(time.Now())) {
+ censusOnce.Do(func() { close(census) })
+ <-ctx.Done()
+ return nil, "", ctx.Err()
+ }
+ return store.Store.Get(ctx, key)
+ }
+ store.put = func(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ out, err := store.Store.Put(ctx, key, body, etag)
+ if strings.HasPrefix(key, "usage/storage/") && err == nil {
+ publishOnce.Do(func() { close(published) })
+ }
+ return out, err
+ }
+ fx.srv.cfg.Releases = store
+ done := make(chan struct{})
+ go func() { defer close(done); fx.srv.StartEconomicsPass(ctx) }()
+ select {
+ case <-census:
+ case <-time.After(5 * time.Second):
+ t.Fatal("census did not enter")
+ }
+ select {
+ case <-published:
+ case <-time.After(5 * time.Second):
+ t.Fatal("census blocked storage worker")
+ }
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("economics scheduler did not stop")
+ }
+}
+
+func TestStorageRetryRejectsCanceledSuccessfulCollection(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 30, 0, 0, time.UTC)
+ _, _, runner := storageRetryFixture(t, &now)
+ published := false
+ runner.publish = func(context.Context, string, []byte) error { published = true; return nil }
+ runner.limit = func(parent context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+ ctx, cancel := context.WithCancel(parent)
+ cancel()
+ return ctx, cancel
+ }
+ runner.collect = func(context.Context, string) (*usage.StorageDay, error) {
+ return &usage.StorageDay{}, nil
+ }
+ runner.wait = func(context.Context, time.Duration) bool { return false }
+ runner.run(t.Context())
+ if published {
+ t.Fatal("canceled collection was published")
+ }
+}
+
+func TestStorageRetrySchedulerJoinsWorker(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.ceCosts = func(context.Context, time.Time, time.Time) (map[string]float64, error) {
+ return nil, io.ErrUnexpectedEOF
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ entered, releaseWorker := make(chan struct{}), make(chan struct{})
+ defer close(releaseWorker)
+ var once sync.Once
+ store := &storageRetryTestStore{Store: fx.rel}
+ store.get = func(ctx context.Context, key string) ([]byte, string, error) {
+ if key == usage.StorageDayKey(usage.DayOf(time.Now())) {
+ once.Do(func() { close(entered) })
+ <-ctx.Done()
+ <-releaseWorker
+ return nil, "", ctx.Err()
+ }
+ if key == usage.CensusDayKey(usage.DayOf(time.Now())) {
+ <-ctx.Done()
+ return nil, "", ctx.Err()
+ }
+ return store.Store.Get(ctx, key)
+ }
+ fx.srv.cfg.Releases = store
+ done := make(chan struct{})
+ go func() { defer close(done); fx.srv.StartEconomicsPass(ctx) }()
+ select {
+ case <-entered:
+ case <-time.After(5 * time.Second):
+ t.Fatal("worker did not enter")
+ }
+ cancel()
+ select {
+ case <-done:
+ t.Fatal("scheduler returned while worker still owns the store")
+ case <-time.After(50 * time.Millisecond):
+ }
+ releaseWorker <- struct{}{}
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("scheduler did not join finished worker")
+ }
+}