amadan

carlos / platform Public

Bound storage sweeps by midnight and verify conflicting publications

claude fb0c6dfc23187d087900d0b687fe8cf715f79a36 parent 1dea9ca
2 files changed, +66 −6
  • internal/console/economicstorageretry.go +4 −3
  • internal/console/economicstorageretry_test.go +62 −3

CI passed — run details

diff --git a/internal/console/economicstorageretry.go b/internal/console/economicstorageretry.go
index 7bcb7b05d..40fadcec0 100644
--- a/internal/console/economicstorageretry.go
+++ b/internal/console/economicstorageretry.go
@@ -41,9 +41,10 @@ func (s *Server) storageRetryRunner() storageRetryRunner {
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, "")
+ key := usage.StorageDayKey(date)
+ _, err := s.cfg.Releases.Put(ctx, key, body, "")
if errors.Is(err, release.ErrCASConflict) {
- return nil
+ _, _, err = s.cfg.Releases.Get(ctx, key)
}
return err
},
@@ -69,7 +70,7 @@ func (r storageRetryRunner) run(ctx context.Context) {
continue
}
date = today
- opCtx, cancel := r.limit(ctx, min(storageOperationTimeout, end.Sub(now)))
+ opCtx, cancel := r.limit(ctx, end.Sub(now))
day, err := r.collect(opCtx, date)
if err == nil {
err = opCtx.Err()
diff --git a/internal/console/economicstorageretry_test.go b/internal/console/economicstorageretry_test.go
index 30e13cd2a..3b3a155ff 100644
--- a/internal/console/economicstorageretry_test.go
+++ b/internal/console/economicstorageretry_test.go
@@ -63,6 +63,63 @@ func storageRetryFixture(t *testing.T, now *time.Time) (*consoleFixture, *storag
return fx, store, runner
}
+func TestStorageRetryCollectionCanExceedPublicationTimeout(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 30, 0, 0, time.UTC)
+ fx, _, runner := storageRetryFixture(t, &now)
+ var deadline time.Time
+ var expire context.CancelFunc
+ runner.limit = func(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+ deadline = now.Add(d)
+ limited, cancel := context.WithCancel(ctx)
+ expire = cancel
+ return limited, cancel
+ }
+ collect := runner.collect
+ runner.collect = func(ctx context.Context, date string) (*usage.StorageDay, error) {
+ day, err := collect(ctx, date)
+ now = now.Add(3 * time.Minute)
+ if !now.Before(deadline) {
+ expire()
+ }
+ return day, err
+ }
+ runner.wait = func(context.Context, time.Duration) bool { return false }
+ runner.run(t.Context())
+ if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03")); err != nil {
+ t.Fatalf("completed same-day sweep was discarded: %v", err)
+ }
+}
+
+func TestStorageRetryConflictRequiresExistingDay(t *testing.T) {
+ now := time.Date(2026, 10, 3, 23, 30, 0, 0, time.UTC)
+ fx, store, runner := storageRetryFixture(t, &now)
+ var bodies [][]byte
+ store.put = func(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ bodies = append(bodies, append([]byte(nil), body...))
+ if len(bodies) == 1 {
+ return "", release.ErrCASConflict
+ }
+ return store.Store.Put(ctx, key, body, etag)
+ }
+ waits := 0
+ runner.wait = func(_ context.Context, d time.Duration) bool {
+ waits++
+ if waits > 1 {
+ return false
+ }
+ now = now.Add(d)
+ return true
+ }
+ runner.run(t.Context())
+ if len(bodies) != 2 || string(bodies[0]) != string(bodies[1]) {
+ t.Fatalf("missing day after conflict was not retried unchanged: %d writes", len(bodies))
+ }
+ raw, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03"))
+ if err != nil || string(raw) != string(bodies[0]) {
+ t.Fatalf("publication did not survive a conflicting writer that failed: %v", err)
+ }
+}
+
func TestStorageRetryWakesAtFinalHour(t *testing.T) {
now := time.Date(2026, 10, 3, 22, 30, 0, 0, time.UTC)
fx, store, runner := storageRetryFixture(t, &now)
@@ -283,6 +340,7 @@ func TestStorageRetryCollectionCancellation(t *testing.T) {
entered := make(chan struct{})
opCancel := make(chan context.CancelFunc, 1)
var once sync.Once
+ var limits []time.Duration
store.dirs = func(ctx context.Context, prefix string) ([]string, error) {
blocked := false
once.Do(func() { blocked = true; close(entered) })
@@ -293,9 +351,7 @@ func TestStorageRetryCollectionCancellation(t *testing.T) {
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)
- }
+ limits = append(limits, d)
op, stop := context.WithCancel(parent)
select {
case opCancel <- stop:
@@ -319,6 +375,9 @@ func TestStorageRetryCollectionCancellation(t *testing.T) {
case <-time.After(5 * time.Second):
t.Fatal("canceled collection kept retry owner blocked")
}
+ if !reflect.DeepEqual(limits, []time.Duration{30 * time.Minute, 29 * time.Minute, 2 * time.Minute}) {
+ t.Fatalf("collection/publication deadlines: %v", limits)
+ }
if _, _, err := fx.rel.Get(t.Context(), usage.StorageDayKey("2026-10-03")); err != nil {
t.Fatal(err)
}