Back off export retries and avoid head-of-line blocking

Review of the pending retry list found that windows which can never be
exported (for example a compute that always errors for a window) sat at
the front of the list and used every per-tick attempt, so newer closed
windows were not exported until the bad ones were evicted up to 48 hours
later. They were also recomputed on every tick, and during a storage
outage each tick computed up to four closed windows only to discard them.

Newly closed windows are now attempted before retries, so a window is
always exported on the first tick after it closes, as before. Failed
windows are retried with a per-window backoff of 1, 2, 4, 8 and then at
most every 12 ticks, oldest first, and after a failed retry no further
retries run in that tick. A backlog now drains within 12 ticks of
recovery rather than strictly in window order.

Also: the last-seen window never moves backwards, so a backwards clock
step can't enqueue a window twice; the current window isn't exported once
the controller is stopping; and stopping with windows still pending logs
a warning.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Signed-off-by: peatey <warwick@automatic.systems>

core/pkg/exporter/controller.go

@@ -3,6 +3,7 @@ package exporter
 import (
 	"fmt"
 	"reflect"
+	"slices"
 	"strings"
 	"time"
 
@@ -104,14 +105,29 @@ const (
 	// defaultMaxExportsPerTick caps how many closed windows are exported in a single tick, so that
 	// draining a backlog after an outage doesn't compute and write every window at once
 	defaultMaxExportsPerTick = 4
+
+	// maxRetryBackoffTicks caps the number of ticks between retries of a failed closed window
+	maxRetryBackoffTicks = 12
 )
 
+// pendingWindow is a closed window awaiting a successful export
+type pendingWindow struct {
+	start time.Time
+	// attempts is the number of failed exports of this window since it closed
+	attempts int
+	// nextTick is the tick at which this window is next due to be retried
+	nextTick uint64
+}
+
 // ComputeExportController[T] is a controller type which leverages a `ComputeSource[T]` and `Exporter[T]`
 // to regularly compute the data for the current resolution and export it on a specific interval.
 //
 // Each tick exports the current (in-progress) window. When a window closes, it is added to a pending
-// list and exported on subsequent ticks until an export succeeds, oldest first. The pending list is
-// bounded; when it overflows, the oldest window is dropped and counted.
+// list and exported on the next tick. A window whose export fails stays pending and is retried with a
+// per-window backoff until an export succeeds. Newly closed windows are exported before retries, and
+// retries go oldest first; after a failed retry no further retries are attempted in that tick, so a
+// storage outage costs at most one retry per tick. The pending list is bounded; when it overflows,
+// the oldest window is dropped and counted.
 type ComputeExportController[T any] struct {
 	runState   atomic.AtomicRunState
 	source     ComputeSource[T]
@@ -119,13 +135,15 @@ type ComputeExportController[T any] struct {
 	resolution time.Duration
 	typeName   string
 
-	// lastTickWindow is the start of the current window at the previous tick
+	// tickCount is the number of ticks run
+	tickCount uint64
+	// lastTickWindow is the latest start of the current window seen by a tick
 	lastTickWindow time.Time
-	// pending holds the start times of closed windows awaiting a successful export, ascending
-	pending []time.Time
+	// pending holds closed windows awaiting a successful export, ascending by start
+	pending []*pendingWindow
 	// maxPendingWindows bounds pending; the oldest windows beyond this are dropped
 	maxPendingWindows int
-	// maxExportsPerTick bounds the closed-window exports attempted per tick
+	// maxExportsPerTick bounds the closed-window exports (first attempts and retries) per tick
 	maxExportsPerTick int
 	// droppedWindows counts closed windows dropped from pending without a successful export
 	droppedWindows uint64
@@ -183,6 +201,9 @@ func (cd *ComputeExportController[T]) Start(interval time.Duration) bool {
 			// if our stop channel receives data, it means we have explicitly called
 			// Stop(), and must reset our AtomicRunState to it's initial idle state
 			case <-cd.runState.OnStop():
+				if n := cd.pendingCount(); n > 0 {
+					log.Warnf("[%s] stopping with %d closed window(s) not yet exported", cd.Name(), n)
+				}
 				cd.runState.Reset()
 				return // exit go routine
 
@@ -197,31 +218,61 @@ func (cd *ComputeExportController[T]) Start(interval time.Duration) bool {
 	return true
 }
 
-// tick runs a single export pass for the provided time: pending closed windows are exported oldest
-// first (up to maxExportsPerTick), followed by the current window.
+// tick runs a single export pass for the provided time: newly closed windows first, then due retries
+// oldest first (together at most maxExportsPerTick), followed by the current window.
 func (cd *ComputeExportController[T]) tick(now time.Time) {
+	cd.tickCount++
+
 	start := now.Truncate(cd.resolution)
 	cd.enqueueClosedWindows(start)
-	cd.lastTickWindow = start
+	// never move backwards, so a backwards clock step can't enqueue the same window twice
+	if start.After(cd.lastTickWindow) {
+		cd.lastTickWindow = start
+	}
 
 	attempts := 0
-	remaining := make([]time.Time, 0, len(cd.pending))
-	for _, windowStart := range cd.pending {
-		if attempts >= cd.maxExportsPerTick || cd.runState.IsStopping() {
-			remaining = append(remaining, windowStart)
-			continue
-		}
-		attempts++
+	for _, firstAttempt := range []bool{true, false} {
+		for _, pw := range cd.pending {
+			if attempts >= cd.maxExportsPerTick || cd.runState.IsStopping() {
+				break
+			}
+			if (pw.attempts == 0) != firstAttempt || pw.nextTick > cd.tickCount {
+				continue
+			}
+			attempts++
 
-		if !cd.exportAndLog(opencost.NewClosedWindow(windowStart, windowStart.Add(cd.resolution))) {
-			remaining = append(remaining, windowStart)
+			if cd.exportAndLog(opencost.NewClosedWindow(pw.start, pw.start.Add(cd.resolution))) {
+				pw.start = time.Time{} // exported; removed below
+				continue
+			}
+
+			pw.attempts++
+			pw.nextTick = cd.tickCount + retryBackoffTicks(pw.attempts)
+
+			// a failed retry most likely means the next one will fail too (e.g. storage is down); stop
+			// retrying until the next tick rather than recomputing more windows only to discard them
+			if !firstAttempt {
+				break
+			}
 		}
 	}
-	cd.pending = remaining
+	cd.pending = slices.DeleteFunc(cd.pending, func(pw *pendingWindow) bool { return pw.start.IsZero() })
 
+	if cd.runState.IsStopping() {
+		return
+	}
 	cd.exportAndLog(opencost.NewClosedWindow(start, start.Add(cd.resolution)))
 }
 
+// retryBackoffTicks returns the number of ticks to wait before retrying a window that has failed the
+// given number of times: 1, 2, 4, 8, then maxRetryBackoffTicks.
+func retryBackoffTicks(attempts int) uint64 {
+	if attempts > 4 {
+		return maxRetryBackoffTicks
+	}
+	return min(uint64(1)<<(attempts-1), maxRetryBackoffTicks)
+}
+
 // enqueueClosedWindows adds every window that has closed since the previous tick to the pending list,
 // dropping the oldest pending windows if the list exceeds maxPendingWindows.
 func (cd *ComputeExportController[T]) enqueueClosedWindows(currentStart time.Time) {
@@ -242,19 +293,19 @@ func (cd *ComputeExportController[T]) enqueueClosedWindows(currentStart time.Tim
 	}
 
 	for ws := first; ws.Before(currentStart); ws = ws.Add(cd.resolution) {
-		cd.pending = append(cd.pending, ws)
+		cd.pending = append(cd.pending, &pendingWindow{start: ws})
 	}
 
 	if over := len(cd.pending) - cd.maxPendingWindows; over > 0 {
-		cd.drop(cd.pending[0], cd.pending[over-1].Add(cd.resolution), over)
-		cd.pending = append([]time.Time(nil), cd.pending[over:]...)
+		cd.drop(cd.pending[0].start, cd.pending[over-1].start.Add(cd.resolution), over)
+		cd.pending = append([]*pendingWindow(nil), cd.pending[over:]...)
 	}
 }
 
-// drop records count closed windows in [start, end) as dropped without a successful export
+// drop records count closed windows between start and end as dropped without a successful export
 func (cd *ComputeExportController[T]) drop(start, end time.Time, count int) {
 	cd.droppedWindows += uint64(count)
-	log.Errorf("[%s] dropping %d closed window(s) in [%s, %s) that were never exported: pending limit of %d reached",
+	log.Errorf("[%s] dropping %d closed window(s) between %s and %s that were never exported: pending limit of %d reached",
 		cd.Name(), count, start.Format(time.RFC3339), end.Format(time.RFC3339), cd.maxPendingWindows)
 }
 

core/pkg/exporter/controller_pending_test.go

@@ -10,10 +10,12 @@ package exporter
 //	                                    (excludes the current, open window)
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/opencost/opencost/core/pkg/opencost"
+	"github.com/opencost/opencost/core/pkg/source"
 )
 
 func TestComputeExportController_EvictsBeyondMaxPending(t *testing.T) {
@@ -97,23 +99,17 @@ func TestComputeExportController_CapsExportsPerTick(t *testing.T) {
 		}
 	}
 
-	// recovery drains oldest-first: 13:30 → 08,09; 13:35 → 10,11; 13:40 → 12.
-	want := map[time.Time]time.Time{
-		at(8, 0, 0):  recovery,
-		at(9, 0, 0):  recovery,
-		at(10, 0, 0): at(13, 35, 0),
-		at(11, 0, 0): at(13, 35, 0),
-		at(12, 0, 0): at(13, 40, 0),
-	}
-	for w, wantAt := range want {
+	// every pending window drains within maxRetryBackoffTicks ticks of recovery
+	drainedBy := recovery.Add(maxRetryBackoffTicks * 5 * time.Minute)
+	for w := at(8, 0, 0); w.Before(at(13, 0, 0)); w = w.Add(time.Hour) {
 		i := firstPostCloseSuccess(recs, w)
 		if i < 0 {
 			t.Errorf("window %s never got a post-close export", hourWindow(w))
 			continue
 		}
-		if !recs[i].Now.Equal(wantAt) {
-			t.Errorf("window %s finalized at %s, want %s (oldest-first, 2 per tick)",
-				hourWindow(w), recs[i].Now.Format("15:04:05"), wantAt.Format("15:04:05"))
+		if recs[i].Now.After(drainedBy) {
+			t.Errorf("window %s finalized at %s, after the drain bound %s",
+				hourWindow(w), recs[i].Now.Format("15:04:05"), drainedBy.Format("15:04:05"))
 		}
 	}
 
@@ -220,3 +216,86 @@ func TestComputeExportController_StopDuringBacklog(t *testing.T) {
 		t.Errorf("expected the 5 un-attempted windows to remain pending, got %d", c.pendingCount())
 	}
 }
+
+// Windows that always fail must not block newer closed windows from being exported (no head-of-line
+// blocking), and are retried with backoff rather than on every tick.
+func TestComputeExportController_PoisonedWindowsDoNotBlock(t *testing.T) {
+	poisoned := map[time.Time]bool{at(8, 0, 0): true, at(9, 0, 0): true, at(10, 0, 0): true, at(11, 0, 0): true, at(12, 0, 0): true}
+	src := &fakeComputeSource[controllerTestSet]{}
+	exp := &fakeComputeExporter[controllerTestSet]{
+		failIf: func(w opencost.Window, _ time.Time) bool { return poisoned[*w.Start()] },
+	}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+
+	ticks := ticksEvery(at(8, 2, 0), at(20, 2, 0), 5*time.Minute)
+	runTicks(c, exp, ticks, nil)
+	recs := exp.Records()
+
+	// every healthy window gets its final export on the first tick after it closes
+	for w := at(13, 0, 0); w.Before(at(20, 0, 0)); w = w.Add(time.Hour) {
+		i := firstPostCloseSuccess(recs, w)
+		if i < 0 {
+			t.Errorf("healthy window %s was never exported after closing", hourWindow(w))
+			continue
+		}
+		if want := w.Add(time.Hour).Add(2 * time.Minute); !recs[i].Now.Equal(want) {
+			t.Errorf("healthy window %s finalized at %s, want %s", hourWindow(w), recs[i].Now.Format("15:04:05"), want.Format("15:04:05"))
+		}
+	}
+
+	// the first poisoned window is retried with backoff: far fewer attempts than ticks since it closed
+	attempts := 0
+	for _, r := range recs {
+		if r.Start.Equal(at(8, 0, 0)) && r.postClose() {
+			attempts++
+		}
+	}
+	ticksSinceClose := len(ticksEvery(at(9, 2, 0), at(20, 2, 0), 5*time.Minute))
+	if attempts == 0 || attempts > ticksSinceClose/maxRetryBackoffTicks+5 {
+		t.Errorf("poisoned window attempted %d times over %d ticks, want backoff", attempts, ticksSinceClose)
+	}
+	if t.Failed() {
+		dumpRecords(t, recs)
+	}
+}
+
+// A backwards clock step followed by a forward one must not enqueue the same window twice.
+func TestComputeExportController_ClockStepDoesNotDuplicatePending(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	exp := &fakeComputeExporter[controllerTestSet]{
+		failIf: func(w opencost.Window, _ time.Time) bool { return w.Start().Equal(at(9, 0, 0)) },
+	}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+
+	runTicks(c, exp, []time.Time{at(9, 30, 0), at(10, 30, 0), at(9, 45, 0), at(10, 35, 0)}, nil)
+	if n := c.pendingCount(); n != 1 {
+		t.Errorf("pendingCount() = %d, want 1", n)
+	}
+}
+
+// A compute that fails with a QueryErrorCollection keeps the window pending until it succeeds.
+func TestComputeExportController_ErrorCollectionKeepsWindowPending(t *testing.T) {
+	failing := true
+	src := &fakeComputeSource[controllerTestSet]{
+		computeFn: func(start, _ time.Time, _ int) (*controllerTestSet, error) {
+			if failing && start.Equal(at(9, 0, 0)) {
+				errs := &source.QueryErrorCollector{}
+				errs.AppendError(&source.QueryError{Query: "q", Error: fmt.Errorf("boom")})
+				return nil, errs
+			}
+			return &controllerTestSet{}, nil
+		},
+	}
+	exp := &fakeComputeExporter[controllerTestSet]{}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+
+	runTicks(c, exp, []time.Time{at(9, 30, 0), at(10, 5, 0)}, nil)
+	if n := c.pendingCount(); n != 1 {
+		t.Fatalf("pendingCount() = %d after error collection, want 1", n)
+	}
+	failing = false
+	runTicks(c, exp, []time.Time{at(10, 10, 0)}, nil)
+	if n := c.pendingCount(); n != 0 {
+		t.Errorf("pendingCount() = %d after recovery, want 0", n)
+	}
+}

core/pkg/exporter/controller_test.go

@@ -227,10 +227,10 @@ func TestComputeExportController_RetriesFailedClosedWindow(t *testing.T) {
 }
 
 // ---------------------------------------------------------------------------
-// Outage: every closed window drains, in order, exactly once
+// Outage: every closed window drains exactly once, within the retry backoff cap
 // ---------------------------------------------------------------------------
 
-func TestComputeExportController_OutageDrainsInOrder(t *testing.T) {
+func TestComputeExportController_OutageDrains(t *testing.T) {
 	src := &fakeComputeSource[controllerTestSet]{}
 	outageStart, outageEnd := at(9, 0, 0), at(14, 0, 0)
 	exp := &fakeComputeExporter[controllerTestSet]{
@@ -262,9 +262,9 @@ func TestComputeExportController_OutageDrainsInOrder(t *testing.T) {
 		}
 	}
 
-	// 2. final exports occur in ascending window order, and 3. a closed window
-	// is never exported again after its first post-close success.
-	var order []time.Time
+	// 2. every final export happens within maxRetryBackoffTicks ticks of recovery, and 3. a closed
+	// window is never exported again after its first post-close success.
+	drainedBy := outageEnd.Add(maxRetryBackoffTicks * 5 * time.Minute)
 	done := map[time.Time]bool{}
 	for _, r := range recs {
 		if !r.postClose() {
@@ -277,16 +277,29 @@ func TestComputeExportController_OutageDrainsInOrder(t *testing.T) {
 		}
 		if r.Success {
 			done[r.Start] = true
-			order = append(order, r.Start)
+			if r.Start.Before(outageEnd) && r.Now.After(drainedBy) {
+				fail("window %s finalized at %s, after the drain bound %s", hourWindow(r.Start), r.Now.Format("15:04:05"), drainedBy.Format("15:04:05"))
+			}
 		}
 	}
-	for i := 1; i < len(order); i++ {
-		if !order[i].After(order[i-1]) {
-			fail("final exports out of order: %s finalized after %s", hourWindow(order[i]), hourWindow(order[i-1]))
+
+	// 4. during the outage, at most one retry (plus any newly closed window) is attempted per tick.
+	for i, now := range ticks {
+		if !now.Before(outageEnd) || now.Before(outageStart) {
+			continue
+		}
+		retries := 0
+		for _, r := range recs {
+			if r.Tick == i && r.postClose() && !r.Start.Equal(now.Truncate(time.Hour).Add(-time.Hour)) {
+				retries++
+			}
+		}
+		if retries > 1 {
+			fail("tick %s attempted %d retries during the outage, want at most 1", now.Format("15:04:05"), retries)
 		}
 	}
 
-	// 4. the current window is attempted on every tick, including during the outage.
+	// 5. the current window is attempted on every tick, including during the outage.
 	for i, now := range ticks {
 		cur := now.Truncate(time.Hour)
 		found := false