cost-model/core/pkg/exporter/controller_pending_test.go

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 (
	"fmt"
	"testing"
	"time"

	"github.com/opencost/opencost/core/pkg/opencost"
	"github.com/opencost/opencost/core/pkg/source"
)

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)
		}
	}

	// 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.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"))
		}
	}

	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())
	}
}

// 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)
	}
}