mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 11:21:49 +00:00
fix: run the reconciler tests in parallel
Each test starts its own embedded JetStream server and bucket, so they were already independent; running them serially just stacked up the fixed settle sleeps. 13.3s to 2.8s under -race.
This commit is contained in:
@@ -62,6 +62,7 @@ func (l *keyLog) stillAt(t *testing.T, n int, why string) {
|
||||
}
|
||||
|
||||
func TestRun_AnUpdateReconcilesTheKeyThatChanged(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-one")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
@@ -84,6 +85,7 @@ func TestRun_AnUpdateReconcilesTheKeyThatChanged(t *testing.T) {
|
||||
// The burst case the whole-set path collapses into one pass becomes one pass
|
||||
// per distinct key here, which is the entire point of the extension.
|
||||
func TestRun_ABurstReconcilesEachKeyOnce(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-burst")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
@@ -111,6 +113,7 @@ func TestRun_ABurstReconcilesEachKeyOnce(t *testing.T) {
|
||||
// key, not five: a fresh read makes the reconcile idempotent, so the four
|
||||
// earlier revisions have nothing left to say.
|
||||
func TestRun_RepeatedWritesToOneKeyReconcileItOnce(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-dedup")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
@@ -136,6 +139,7 @@ func TestRun_RepeatedWritesToOneKeyReconcileItOnce(t *testing.T) {
|
||||
// pending, so a change arriving while it is being reconciled is a new piece of
|
||||
// work rather than one the running pass is assumed to cover.
|
||||
func TestRun_AChangeDuringAKeysReconcileIsNotLost(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-inflight")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
keys.block = make(chan struct{})
|
||||
@@ -164,6 +168,7 @@ func TestRun_AChangeDuringAKeysReconcileIsNotLost(t *testing.T) {
|
||||
// KeyFor exists for a caller whose unit of work is coarser than a key. Two
|
||||
// instances in one account are one recompute, not two.
|
||||
func TestRun_KeyForMapsUpdatesOntoTheCallersUnitOfWork(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-keyfor")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
@@ -193,6 +198,7 @@ func TestRun_KeyForMapsUpdatesOntoTheCallersUnitOfWork(t *testing.T) {
|
||||
// lived in a value the tombstone does not carry — falls back to the pass that
|
||||
// needs no attribution.
|
||||
func TestRun_AnUnattributableUpdateFallsBackToTheWholeSet(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-fallback")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
@@ -224,6 +230,7 @@ func TestRun_AnUnattributableUpdateFallsBackToTheWholeSet(t *testing.T) {
|
||||
// replacement: a key the queue never learned about — deleted while the watch
|
||||
// was down, so nothing observed it — is still covered.
|
||||
func TestRun_TheResyncStillRunsTheWholeSet(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-resync")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
@@ -240,6 +247,7 @@ func TestRun_TheResyncStillRunsTheWholeSet(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestRun_AFailedKeyDoesNotStopTheLoop(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "perkey-failure")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
|
||||
@@ -102,6 +102,7 @@ func run(t *testing.T, cfg reconciler.Config) {
|
||||
}
|
||||
|
||||
func TestRun_ReconcilesOnceAtStartup(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-startup")
|
||||
p := newPasses()
|
||||
|
||||
@@ -117,6 +118,7 @@ func TestRun_ReconcilesOnceAtStartup(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestRun_AnUpdateTriggersAPass(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-update")
|
||||
p := newPasses()
|
||||
|
||||
@@ -136,6 +138,7 @@ func TestRun_AnUpdateTriggersAPass(t *testing.T) {
|
||||
// multi-key write is one logical change, and a full-recompute reconcile gains
|
||||
// nothing from running once per key.
|
||||
func TestRun_ABurstCollapsesIntoOnePass(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-burst")
|
||||
p := newPasses()
|
||||
|
||||
@@ -157,6 +160,7 @@ func TestRun_ABurstCollapsesIntoOnePass(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestRun_ResyncFiresWithNoWatchTraffic(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-resync")
|
||||
p := newPasses()
|
||||
|
||||
@@ -173,6 +177,7 @@ func TestRun_ResyncFiresWithNoWatchTraffic(t *testing.T) {
|
||||
// TestRun_NoSourcesStillReconcilesOnTheResync pins the degenerate case: a loop
|
||||
// with nothing watchable is the pre-watch behaviour, not a dead loop.
|
||||
func TestRun_NoSourcesStillReconcilesOnTheResync(t *testing.T) {
|
||||
t.Parallel()
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
@@ -187,6 +192,7 @@ func TestRun_NoSourcesStillReconcilesOnTheResync(t *testing.T) {
|
||||
// TestRun_AFailedPassDoesNotStopTheLoop covers the reason Reconcile's error is
|
||||
// logged rather than returned: the loop outlives any single pass.
|
||||
func TestRun_AFailedPassDoesNotStopTheLoop(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-failure")
|
||||
p := newPasses()
|
||||
var calls atomic.Int64
|
||||
@@ -211,6 +217,7 @@ func TestRun_AFailedPassDoesNotStopTheLoop(t *testing.T) {
|
||||
// waking a loop that does not care about it — the ELBv2 bucket holds listeners
|
||||
// and rules alongside the load balancers DNS reconciles on.
|
||||
func TestRun_AFilterExcludesOtherKeys(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-filter")
|
||||
p := newPasses()
|
||||
|
||||
@@ -252,6 +259,7 @@ func (s *bucketSet) add(b *kvstore.Bucket) {
|
||||
// case: EKS and RDS create a bucket per account, and JetStream has no
|
||||
// bucket-created event, so discovery rides on the resync.
|
||||
func TestRun_ABucketAppearingAfterStartupIsPickedUp(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
js := testutil.NewJetStream(t, nc)
|
||||
first := kvstore.New[record](js, kvstore.Config{Name: "reconciler-acct-a", History: 1})
|
||||
@@ -286,6 +294,7 @@ func TestRun_ABucketAppearingAfterStartupIsPickedUp(t *testing.T) {
|
||||
// JetStream behind it cannot be watched, and the loop must degrade to the timer
|
||||
// rather than stop.
|
||||
func TestRun_AnUnwatchableSourceStillResyncs(t *testing.T) {
|
||||
t.Parallel()
|
||||
unwatchable := kvstore.NewBucket(nil, kvstore.Config{
|
||||
Name: "reconciler-unwatchable",
|
||||
Missing: "no JetStream configured",
|
||||
@@ -303,6 +312,7 @@ func TestRun_AnUnwatchableSourceStillResyncs(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestFixed_ReportsItsBucketAndFilter(t *testing.T) {
|
||||
t.Parallel()
|
||||
bucket := kvstore.NewBucket(nil, kvstore.Config{Name: "b"})
|
||||
src := reconciler.Fixed(bucket, "node.*")
|
||||
|
||||
@@ -314,6 +324,7 @@ func TestFixed_ReportsItsBucketAndFilter(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestBucket_WatchOnANilJetStreamReportsTheConfiguredMessage(t *testing.T) {
|
||||
t.Parallel()
|
||||
bucket := kvstore.NewBucket(nil, kvstore.Config{
|
||||
Name: "b",
|
||||
Missing: "watch test: no JetStream client configured",
|
||||
@@ -327,6 +338,7 @@ func TestBucket_WatchOnANilJetStreamReportsTheConfiguredMessage(t *testing.T) {
|
||||
// watcher established over a bucket that already holds keys must not replay
|
||||
// them, or every re-establish would fire a redundant pass per existing key.
|
||||
func TestBucket_WatchDeliversUpdatesOnly(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "reconciler-updatesonly")
|
||||
require.NoError(t, store.Set(t.Context(), "existing", &record{Name: "existing"}))
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
// exists: a loop waiting on something outside KV gets no watch update at all,
|
||||
// so without this it would run only on the resync.
|
||||
func TestRun_ARevisitDeadlineFiresWithNoWatchTraffic(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "revisit-deadline")
|
||||
p := newPasses()
|
||||
|
||||
@@ -32,6 +33,7 @@ func TestRun_ARevisitDeadlineFiresWithNoWatchTraffic(t *testing.T) {
|
||||
// may not move: the resync is the outer bound on staleness, so a longer deadline
|
||||
// is ignored rather than allowed to defer past it.
|
||||
func TestRun_ARevisitDeadlineIsClampedToTheResync(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "revisit-clamped")
|
||||
p := newPasses()
|
||||
|
||||
@@ -49,6 +51,7 @@ func TestRun_ARevisitDeadlineIsClampedToTheResync(t *testing.T) {
|
||||
// finishes: once the last transitional resource settles it stops asking, and the
|
||||
// loop must fall back to the resync rather than keep the old deadline armed.
|
||||
func TestRun_DroppingTheDeadlineStopsTheEarlyPasses(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "revisit-dropped")
|
||||
p := newPasses()
|
||||
var calls atomic.Int64
|
||||
@@ -74,6 +77,7 @@ func TestRun_DroppingTheDeadlineStopsTheEarlyPasses(t *testing.T) {
|
||||
// the resync. A pass that fails partway through a transition is exactly the one
|
||||
// that most needs revisiting on its own schedule.
|
||||
func TestRun_AFailedPassKeepsItsRevisitDeadline(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "revisit-failed")
|
||||
p := newPasses()
|
||||
|
||||
@@ -94,6 +98,7 @@ func TestRun_AFailedPassKeepsItsRevisitDeadline(t *testing.T) {
|
||||
// The wait is capped well below the deadline asked for, so a pass arriving at all
|
||||
// proves the change woke the loop rather than the timer.
|
||||
func TestRun_AChangeBeforeTheDeadlineStillTriggersAPass(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "revisit-change-wins")
|
||||
p := newPasses()
|
||||
|
||||
@@ -113,6 +118,7 @@ func TestRun_AChangeBeforeTheDeadlineStillTriggersAPass(t *testing.T) {
|
||||
// deadline serves. Nothing changed, so there is no key to name, and only a
|
||||
// whole-set pass can work out what the next deadline should be.
|
||||
func TestRun_ADeadlinePassRunsTheWholeSet(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "revisit-whole-set")
|
||||
p, keys := newPasses(), newKeyLog()
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
// than a KV write would otherwise see it only when the next pass happened to
|
||||
// look, which is the poll the watch was supposed to replace.
|
||||
func TestRun_ATriggerWakesTheLoopWithNoWatchTraffic(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "trigger-wakes")
|
||||
p := newPasses()
|
||||
trigger := make(chan struct{}, 1)
|
||||
@@ -35,6 +36,7 @@ func TestRun_ATriggerWakesTheLoopWithNoWatchTraffic(t *testing.T) {
|
||||
// A quiet trigger must not be a source of passes on its own: the loop still
|
||||
// falls back to the resync and nothing else.
|
||||
func TestRun_AQuietTriggerAddsNoPasses(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "trigger-quiet")
|
||||
p := newPasses()
|
||||
trigger := make(chan struct{}, 1)
|
||||
@@ -55,6 +57,7 @@ func TestRun_AQuietTriggerAddsNoPasses(t *testing.T) {
|
||||
// caller signalling several times in a row is describing one thing to converge,
|
||||
// not several.
|
||||
func TestRun_ABurstOfTriggersCoalescesIntoOnePass(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "trigger-burst")
|
||||
p := newPasses()
|
||||
trigger := make(chan struct{}, 8)
|
||||
@@ -80,6 +83,7 @@ func TestRun_ABurstOfTriggersCoalescesIntoOnePass(t *testing.T) {
|
||||
// A trigger names no key, so a per-key loop has to be served by the whole-set
|
||||
// pass. Anything else would leave the caller's signal describing nothing.
|
||||
func TestRun_ATriggerRunsTheWholeSetOnAPerKeyLoop(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "trigger-whole-set")
|
||||
whole := newPasses()
|
||||
keyed := newPasses()
|
||||
@@ -104,6 +108,7 @@ func TestRun_ATriggerRunsTheWholeSetOnAPerKeyLoop(t *testing.T) {
|
||||
// The deadline a triggered pass returns has to be armed like any other, or a
|
||||
// loop woken by a trigger would silently lose the schedule it just asked for.
|
||||
func TestRun_ATriggeredPassStillArmsItsDeadline(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "trigger-deadline")
|
||||
p := newPasses()
|
||||
trigger := make(chan struct{}, 1)
|
||||
@@ -125,6 +130,7 @@ func TestRun_ATriggeredPassStillArmsItsDeadline(t *testing.T) {
|
||||
// Closing the trigger is how a caller says its source is finished. The loop must
|
||||
// survive it rather than spin on a closed channel.
|
||||
func TestRun_AClosedTriggerDoesNotSpin(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, _ := newBucket(t, "trigger-closed")
|
||||
p := newPasses()
|
||||
trigger := make(chan struct{})
|
||||
|
||||
Reference in New Issue
Block a user