Retry closed export windows until they are exported

ComputeExportController derived the previous window from the time of the
last successful export, and only ever re-exported that one window. If a
closed window's export failed while the current window's succeeded, the
closed window was never exported again, leaving its object permanently
missing from storage.

Closed windows now go into a bounded pending list and are retried on each
tick, oldest first, until an export succeeds. The current window is still
exported every tick. The list holds 48 windows for sub-daily resolutions
and 7 for daily; overflow drops the oldest with an error log and a
counter. At most 4 closed windows are exported per tick so that draining
a backlog after an outage is spread out, and draining stops early when
the controller is stopped.

Without failures the export cadence is unchanged: each closed window is
exported once, on the first tick after it closes.

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

core/pkg/exporter/controller.go

@@ -94,16 +94,42 @@ func (cd *EventExportController[T]) Stop() {
 	cd.runState.Stop()
 }
 
+const (
+	// defaultMaxPendingWindows is the number of closed sub-daily windows retained for retry
+	defaultMaxPendingWindows = 48
+
+	// defaultMaxPendingDailyWindows is the number of closed daily (or longer) windows retained for retry
+	defaultMaxPendingDailyWindows = 7
+
+	// 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
+)
+
 // 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.
 type ComputeExportController[T any] struct {
 	runState   atomic.AtomicRunState
 	source     ComputeSource[T]
 	exporter   ComputeExporter[T]
 	resolution time.Duration
-	lastExport time.Time
 	typeName   string
 
+	// lastTickWindow is the start of the current window at the previous tick
+	lastTickWindow time.Time
+	// pending holds the start times of closed windows awaiting a successful export, ascending
+	pending []time.Time
+	// maxPendingWindows bounds pending; the oldest windows beyond this are dropped
+	maxPendingWindows int
+	// maxExportsPerTick bounds the closed-window exports attempted per tick
+	maxExportsPerTick int
+	// droppedWindows counts closed windows dropped from pending without a successful export
+	droppedWindows uint64
+
 	// now returns the current time; overridable for tests
 	now func() time.Time
 }
@@ -114,12 +140,19 @@ func NewComputeExportController[T any](
 	exporter ComputeExporter[T],
 	resolution time.Duration,
 ) *ComputeExportController[T] {
+	maxPending := defaultMaxPendingWindows
+	if resolution >= timeutil.Day {
+		maxPending = defaultMaxPendingDailyWindows
+	}
+
 	return &ComputeExportController[T]{
-		source:     source,
-		resolution: resolution,
-		exporter:   exporter,
-		typeName:   reflect.TypeFor[T]().String(),
-		now:        func() time.Time { return time.Now().UTC() },
+		source:            source,
+		resolution:        resolution,
+		exporter:          exporter,
+		typeName:          reflect.TypeFor[T]().String(),
+		maxPendingWindows: maxPending,
+		maxExportsPerTick: defaultMaxExportsPerTick,
+		now:               func() time.Time { return time.Now().UTC() },
 	}
 }
 
@@ -164,58 +197,92 @@ func (cd *ComputeExportController[T]) Start(interval time.Duration) bool {
 	return true
 }
 
-// tick runs a single export pass for the provided time.
+// 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.
 func (cd *ComputeExportController[T]) tick(now time.Time) {
-	windows := cd.exportWindowsFor(now)
-
-	for _, window := range windows {
-		err := cd.export(window)
-		if err != nil {
-			// Check ErrorCollection to set Warnings and Errors
-			if source.IsErrorCollection(err) {
-				c := err.(source.QueryErrorCollection)
-				errors, warnings := c.ToErrorAndWarningStrings()
-
-				cd.logErrors(window, warnings, errors)
-				continue
-			}
+	start := now.Truncate(cd.resolution)
+	cd.enqueueClosedWindows(start)
+	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++
 
-			log.Errorf("[%s] %s", cd.typeName, err)
-		} else {
-			cd.lastExport = now
+		if !cd.exportAndLog(opencost.NewClosedWindow(windowStart, windowStart.Add(cd.resolution))) {
+			remaining = append(remaining, windowStart)
 		}
 	}
+	cd.pending = remaining
+
+	cd.exportAndLog(opencost.NewClosedWindow(start, start.Add(cd.resolution)))
 }
 
-// exportWindows uses the last export time to determine the current time windows to
-// export. This will, at most, return 2 windows: the previous resolution window and
-// the current resolution window.
-func (cd *ComputeExportController[T]) exportWindowsFor(now time.Time) []opencost.Window {
-	start := now.Truncate(cd.resolution)
-	end := start.Add(cd.resolution)
+// 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) {
+	// on the first tick there is no previous window; on a backwards clock step nothing has closed
+	if cd.lastTickWindow.IsZero() || !currentStart.After(cd.lastTickWindow) {
+		return
+	}
 
-	if cd.lastExport.IsZero() {
-		return []opencost.Window{
-			opencost.NewClosedWindow(start, end),
-		}
+	first := cd.lastTickWindow
+	closed := int(currentStart.Sub(first) / cd.resolution)
+
+	// if more windows closed than can be retained (e.g. a long stall), skip straight to the ones we
+	// can keep rather than enqueueing and evicting each one
+	if closed > cd.maxPendingWindows {
+		skipped := closed - cd.maxPendingWindows
+		cd.drop(first, first.Add(time.Duration(skipped)*cd.resolution), skipped)
+		first = first.Add(time.Duration(skipped) * cd.resolution)
 	}
 
-	lastStart := cd.lastExport.Truncate(cd.resolution)
-	if lastStart.Equal(start) {
-		return []opencost.Window{
-			opencost.NewClosedWindow(start, end),
-		}
+	for ws := first; ws.Before(currentStart); ws = ws.Add(cd.resolution) {
+		cd.pending = append(cd.pending, ws)
 	}
-	lastEnd := lastStart.Add(cd.resolution)
 
-	// we've identified that the last export window is not the same as the current,
-	// so we should export the previous resolution window as well as the current one
-	return []opencost.Window{
-		opencost.NewClosedWindow(lastStart, lastEnd),
-		opencost.NewClosedWindow(start, end),
+	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:]...)
 	}
 }
 
+// drop records count closed windows in [start, 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",
+		cd.Name(), count, start.Format(time.RFC3339), end.Format(time.RFC3339), cd.maxPendingWindows)
+}
+
+// pendingCount returns the number of closed windows awaiting a successful export
+func (cd *ComputeExportController[T]) pendingCount() int {
+	return len(cd.pending)
+}
+
+// exportAndLog exports the window, logging any error, and returns true on success
+func (cd *ComputeExportController[T]) exportAndLog(window opencost.Window) bool {
+	err := cd.export(window)
+	if err == nil {
+		return true
+	}
+
+	// Check ErrorCollection to set Warnings and Errors
+	if source.IsErrorCollection(err) {
+		c := err.(source.QueryErrorCollection)
+		errors, warnings := c.ToErrorAndWarningStrings()
+
+		cd.logErrors(window, warnings, errors)
+		return false
+	}
+
+	log.Errorf("[%s] %s", cd.typeName, err)
+	return false
+}
+
 // export computes and exports the data for a given time window
 func (cd *ComputeExportController[T]) export(window opencost.Window) error {
 	if window.IsOpen() {

core/pkg/exporter/controller_pending_test.go

@@ -0,0 +1,222 @@
+package exporter
+
+// These tests exercise the bounded pending set introduced by OC-01. They rely
+// on unexported controller members added by the fix:
+//
+//	maxPendingWindows  (integer field)  cap on closed windows awaiting export
+//	maxExportsPerTick  (integer field)  cap on closed-window exports per tick
+//	droppedWindows     (uint64 field)   count of windows evicted from pending
+//	pendingCount() int                  closed windows currently pending
+//	                                    (excludes the current, open window)
+
+import (
+	"testing"
+	"time"
+
+	"github.com/opencost/opencost/core/pkg/opencost"
+)
+
+func TestComputeExportController_EvictsBeyondMaxPending(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	// 6 closed windows fall in the outage: [08:00..13:00] (all close in [09:00,14:30)).
+	outageStart, outageEnd := at(9, 0, 0), at(14, 30, 0)
+	exp := &fakeComputeExporter[controllerTestSet]{
+		failIf: func(_ opencost.Window, now time.Time) bool {
+			return !now.Before(outageStart) && now.Before(outageEnd)
+		},
+	}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+	c.maxPendingWindows = 3
+
+	ticks := ticksEvery(at(8, 0, 0), at(16, 30, 0), 5*time.Minute)
+	runTicks(c, exp, ticks, func(_ int, now time.Time) {
+		if n := c.pendingCount(); n > 3 {
+			t.Errorf("after tick %s pendingCount() = %d, want <= 3", now.Format("15:04:05"), n)
+		}
+	})
+	recs := exp.Records()
+
+	evicted := []time.Time{at(8, 0, 0), at(9, 0, 0), at(10, 0, 0)}
+	kept := []time.Time{at(11, 0, 0), at(12, 0, 0), at(13, 0, 0)}
+
+	for _, w := range kept {
+		if firstPostCloseSuccess(recs, w) < 0 {
+			t.Errorf("window %s is within maxPendingWindows but never got a post-close export", hourWindow(w))
+		}
+	}
+	for _, w := range evicted {
+		if i := firstPostCloseSuccess(recs, w); i >= 0 {
+			t.Errorf("window %s should have been evicted but was exported post-close at %s",
+				hourWindow(w), recs[i].Now.Format("15:04:05"))
+		}
+	}
+	if got := uint64(c.droppedWindows); got != uint64(len(evicted)) {
+		t.Errorf("droppedWindows = %d, want %d", got, len(evicted))
+	}
+
+	if t.Failed() {
+		dumpRecords(t, recs)
+	}
+}
+
+func TestComputeExportController_CapsExportsPerTick(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	// outage covers closes of [08:00..12:00] → 5 closed windows pending at 13:30.
+	outageStart, recovery := at(9, 0, 0), at(13, 30, 0)
+	exp := &fakeComputeExporter[controllerTestSet]{
+		failIf: func(_ opencost.Window, now time.Time) bool {
+			return !now.Before(outageStart) && now.Before(recovery)
+		},
+	}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+	c.maxExportsPerTick = 2
+
+	ticks := ticksEvery(at(8, 0, 0), at(15, 0, 0), 5*time.Minute)
+	runTicks(c, exp, ticks, nil)
+	recs := exp.Records()
+
+	// per tick: at most 2 closed-window attempts, plus the current window.
+	for i, now := range ticks {
+		cur := now.Truncate(time.Hour)
+		closed, current := 0, 0
+		for _, r := range recs {
+			if r.Tick != i {
+				continue
+			}
+			if r.Start.Equal(cur) {
+				current++
+			} else {
+				closed++
+			}
+		}
+		if closed > 2 {
+			t.Errorf("tick %s attempted %d closed windows, want <= 2", now.Format("15:04:05"), closed)
+		}
+		if current != 1 {
+			t.Errorf("tick %s attempted the current window %d times, want 1", now.Format("15:04:05"), current)
+		}
+	}
+
+	// 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 {
+		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 t.Failed() {
+		dumpRecords(t, recs)
+	}
+}
+
+// Without failures, every closed window gets exactly one post-close export, on the first tick after it
+// closes: the same cadence as before pending retries existed.
+func TestComputeExportController_NoFailuresExportsEachClosedWindowOnce(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	exp := &fakeComputeExporter[controllerTestSet]{}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+
+	ticks := ticksEvery(at(8, 2, 0), at(12, 2, 0), 5*time.Minute)
+	runTicks(c, exp, ticks, nil)
+
+	for w := at(8, 0, 0); w.Before(at(12, 0, 0)); w = w.Add(time.Hour) {
+		var postClose []exportRecord[controllerTestSet]
+		for _, r := range exp.Records() {
+			if r.Start.Equal(w) && r.postClose() {
+				postClose = append(postClose, r)
+			}
+		}
+		if len(postClose) != 1 {
+			t.Fatalf("window %s got %d post-close exports, want 1", hourWindow(w), len(postClose))
+		}
+		if want := w.Add(time.Hour).Add(2 * time.Minute); !postClose[0].Now.Equal(want) {
+			t.Errorf("window %s exported post-close at %s, want first tick after close %s", hourWindow(w), postClose[0].Now.Format("15:04:05"), want.Format("15:04:05"))
+		}
+	}
+	if c.pendingCount() != 0 || c.droppedWindows != 0 {
+		t.Errorf("expected nothing pending or dropped, got pending=%d dropped=%d", c.pendingCount(), c.droppedWindows)
+	}
+}
+
+// A tick that arrives after many windows have closed (a long stall or a clock jump) enqueues only the
+// most recent maxPendingWindows and counts the rest as dropped, without iterating each one.
+func TestComputeExportController_StallBeyondMaxPending(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	exp := &fakeComputeExporter[controllerTestSet]{}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+	c.maxPendingWindows = 3
+	c.maxExportsPerTick = 10
+
+	runTicks(c, exp, []time.Time{at(0, 30, 0), at(10, 30, 0)}, nil)
+
+	// windows 00:00..09:00 closed between ticks: 07, 08 and 09 are retained and exported, 7 are dropped
+	if c.droppedWindows != 7 {
+		t.Errorf("droppedWindows = %d, want 7", c.droppedWindows)
+	}
+	for w := at(0, 0, 0); w.Before(at(10, 0, 0)); w = w.Add(time.Hour) {
+		exported := firstPostCloseSuccess(exp.Records(), w) >= 0
+		if want := !w.Before(at(7, 0, 0)); exported != want {
+			t.Errorf("window %s post-close exported = %v, want %v", hourWindow(w), exported, want)
+		}
+	}
+}
+
+// A clock that steps backwards enqueues nothing and does not panic.
+func TestComputeExportController_ClockStepsBackwards(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	exp := &fakeComputeExporter[controllerTestSet]{}
+	c := NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+
+	runTicks(c, exp, []time.Time{at(10, 30, 0), at(9, 30, 0), at(10, 5, 0)}, nil)
+	if c.pendingCount() != 0 || c.droppedWindows != 0 {
+		t.Errorf("expected nothing pending or dropped, got pending=%d dropped=%d", c.pendingCount(), c.droppedWindows)
+	}
+}
+
+// Stop() during a backlog stops draining closed windows within the current tick.
+func TestComputeExportController_StopDuringBacklog(t *testing.T) {
+	src := &fakeComputeSource[controllerTestSet]{}
+	outageEnd := at(14, 0, 0)
+	var c *ComputeExportController[controllerTestSet]
+	exp := &fakeComputeExporter[controllerTestSet]{}
+	exp.failIf = func(w opencost.Window, now time.Time) bool {
+		if now.Equal(outageEnd) && !w.Start().Equal(outageEnd) {
+			// first closed-window export of the recovery tick: stop the controller
+			c.runState.Stop()
+		}
+		return now.Before(outageEnd)
+	}
+	c = NewComputeExportController[controllerTestSet](src, exp, time.Hour)
+	c.maxExportsPerTick = 10
+	if !c.runState.Start() {
+		t.Fatalf("failed to start run state")
+	}
+
+	runTicks(c, exp, ticksEvery(at(8, 30, 0), outageEnd, 30*time.Minute), nil)
+
+	closedAttempts := 0
+	for _, r := range exp.Records() {
+		if r.Now.Equal(outageEnd) && !r.Start.Equal(outageEnd) {
+			closedAttempts++
+		}
+	}
+	if closedAttempts != 1 {
+		t.Errorf("expected draining to stop after the first closed window once stopped, got %d attempts", closedAttempts)
+	}
+	if c.pendingCount() != 5 {
+		t.Errorf("expected the 5 un-attempted windows to remain pending, got %d", c.pendingCount())
+	}
+}