internal/agent/reconcile/worker_test.go
Ref: Size: 13.2 KiB History
package reconcile
import (
"context"
"sync"
"testing"
"time"
"github.com/a73x/eitri/internal/pb"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// wedge returns an Images func that blocks every fetch until release is closed,
// and signals arrived once per call so a test can observe how many VMs are
// inside a slow operation at the same moment.
func wedge(arrived chan<- struct{}, release <-chan struct{}) func(context.Context, string, string, func(int64, int64)) (string, error) {
return func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
select {
case arrived <- struct{}{}:
default:
}
select {
case <-release:
return "/cache/x.raw", nil
case <-ctx.Done():
return "", ctx.Err()
}
}
}
// stepNoWait drives one tick the way production does: dispatch and take the
// report without waiting on any worker. It is the f.step helper's opposite —
// f.step waits for every worker to go idle, which is exactly the blocking these
// tests exist to prove production does NOT do — and it fails the test if Step
// does not come back promptly. Every test below wedges a VM, so a Step that
// waited on its workers would hang the whole package instead of naming the
// property that broke.
func stepNoWait(t *testing.T, f *fixture, s *pb.Snapshot) *pb.Report {
t.Helper()
done := make(chan *pb.Report, 1)
go func() { done <- f.eng.Step(context.Background(), s) }()
select {
case rep := <-done:
return rep
case <-time.After(2 * time.Second):
t.Fatal("Step blocked on a busy VM: the heartbeat is not protected")
return nil
}
}
// TestBusyVMDoesNotBlockTheReport is the headline property: a VM wedged in a
// slow operation must not delay the host report. This is why the reconcile work
// lives in per-VM workers at all — the heartbeat is what keeps the host online.
func TestBusyVMDoesNotBlockTheReport(t *testing.T) {
f := setup(t)
arrived := make(chan struct{}, 1)
release := make(chan struct{})
f.eng.Images = wedge(arrived, release)
// Dispatch vm1 and wait until its create is actually inside the slow fetch.
stepNoWait(t, f, snap(1, vm("vm1")))
select {
case <-arrived:
case <-time.After(2 * time.Second):
t.Fatal("worker never started the create")
}
// The next tick must come back promptly even though vm1 is still wedged —
// and it carries the wedged VM's row, saying what the wedge is. A create
// publishes as it goes now (see worker.publish), which is precisely the case
// this test wedges: before, a VM's slowest minutes were a report with no row
// for it at all.
rep := stepNoWait(t, f, snap(1, vm("vm1")))
row := findVM(rep, "vm1")
require.NotNil(t, row, "a wedged create must still be in the report")
assert.Equal(t, "creating", row.GetPhase())
assert.Equal(t, "downloading image", row.GetStatusDetail(), "the row says which step is wedged")
// Once the operation completes the VM publishes and appears in the report.
close(release)
f.eng.manager().waitIdle()
assert.Equal(t, "ready", findVM(f.aggregateNow(1), "vm1").GetPhase())
}
// TestVMsReconcileConcurrently pins that one VM's slow operation does not hold
// up another's: under the old single-threaded step, vm2 could not start until
// vm1 finished.
func TestVMsReconcileConcurrently(t *testing.T) {
f := setup(t)
arrived := make(chan struct{}, 2)
release := make(chan struct{})
f.eng.Images = wedge(arrived, release)
stepNoWait(t, f, snap(1, vm("vm1"), vm("vm2")))
for range 2 {
select {
case <-arrived:
case <-time.After(2 * time.Second):
t.Fatal("VMs did not reconcile concurrently: one worker blocked the other")
}
}
close(release)
f.eng.manager().waitIdle()
}
// TestLatestSpecWinsForABusyVM pins coalescing: desired-state updates that
// arrive while a VM is busy overwrite each other, so the worker reconciles
// against the NEWEST desired when it comes free — a superseded intermediate
// state is never acted on.
func TestLatestSpecWinsForABusyVM(t *testing.T) {
f := setup(t)
arrived := make(chan struct{}, 1)
release := make(chan struct{})
f.eng.Images = wedge(arrived, release)
stepNoWait(t, f, snap(1, vm("vm1")))
select {
case <-arrived:
case <-time.After(2 * time.Second):
t.Fatal("worker never started the create")
}
// Both land while vm1 is wedged; the second must overwrite the first.
stepNoWait(t, f, snap(2, vm("vm1", stopped)))
stepNoWait(t, f, snap(3, vm("vm1")))
close(release)
f.eng.manager().waitIdle()
rep := f.aggregateNow(3) // the newest snapshot the worker reconciled against
assert.Equal(t, "running", findVM(rep, "vm1").GetPowerState())
assert.Empty(t, f.prov.shutdown, "the superseded 'stopped' desired must never be reconciled")
}
// TestWorkerLifecycleTracksLiveVMs pins spawn-on-first-sight and reap-when-gone:
// worker count tracks the union of desired state and local records, so a
// destroyed VM leaves no goroutine behind.
func TestWorkerLifecycleTracksLiveVMs(t *testing.T) {
f := setup(t)
m := f.eng.manager()
f.step(snap(1, vm("vm1")))
require.Equal(t, 1, m.count(), "a worker is spawned on first sight")
f.step(snap(2, tombstoned(vm("vm1"))))
assert.Equal(t, 1, m.count(), "quarantined: still has a record, still reconciling")
f.now = f.now.Add(6 * time.Minute) // past TombstoneGrace
f.step(snap(2, tombstoned(vm("vm1"))))
recs, _ := f.st.LoadVMs()
require.NotContains(t, recs, "vm1", "precondition: destroyed")
assert.Equal(t, 1, m.count(), "still in desired (tombstoned), awaiting the destroy ack")
f.step(snap(3))
assert.Equal(t, 0, m.count(), "gone from desired and records: worker reaped")
}
// TestStopEndsEveryWorker pins shutdown: Stop drops every worker so none
// outlives the engine.
func TestStopEndsEveryWorker(t *testing.T) {
f := setup(t)
f.step(snap(1, vm("vm1"), vm("vm2")))
require.Equal(t, 2, f.eng.manager().count())
f.eng.Stop()
assert.Equal(t, 0, f.eng.manager().count())
}
// TestCreateConcurrencyIsCapped pins the create throttle: per-VM workers made
// creates concurrent, and without a cap N simultaneous VMs mean N image fetches
// and N multi-GB disk copies against one device. Three VMs are dispatched under
// a cap of 2, so the third must wait for a slot.
func TestCreateConcurrencyIsCapped(t *testing.T) {
f := setup(t)
f.eng.MaxConcurrentCreates = 2
entered := make(chan struct{}, 3)
release := make(chan struct{})
f.eng.Images = func(ctx context.Context, url, sha string, _ func(done, total int64)) (string, error) {
entered <- struct{}{}
<-release
return "/cache/x.raw", nil
}
f.eng.Step(context.Background(), snap(1, vm("vm1"), vm("vm2"), vm("vm3")))
// Two VMs take the available slots.
for range 2 {
select {
case <-entered:
case <-time.After(2 * time.Second):
t.Fatal("the throttle should admit up to its cap")
}
}
// The third must NOT enter while the first two still hold their slots.
select {
case <-entered:
t.Fatal("a third VM entered the I/O region: the create cap is not enforced")
case <-time.After(250 * time.Millisecond):
}
close(release)
f.eng.manager().waitIdle()
assert.Len(t, f.prov.prepared, 3, "all three VMs still complete, just not at once")
}
// TestCreateConcurrencyZeroIsUnlimited: the zero value must not throttle.
func TestCreateConcurrencyZeroIsUnlimited(t *testing.T) {
f := setup(t)
arrived := make(chan struct{}, 3)
release := make(chan struct{})
f.eng.Images = wedge(arrived, release)
f.eng.Step(context.Background(), snap(1, vm("vm1"), vm("vm2"), vm("vm3")))
for range 3 {
select {
case <-arrived:
case <-time.After(2 * time.Second):
t.Fatal("MaxConcurrentCreates==0 must not throttle")
}
}
close(release)
f.eng.manager().waitIdle()
}
// ---- exclusion observables ----
//
// Every structure in this package is individually mutex-guarded, so an
// execution that runs two passes for one VM performs no unsynchronized access
// and -race stays silent by construction. Exclusion has to be OBSERVED: the
// gauge below is read at the moment two passes would be inside the same
// operation, which is the only moment they differ from one pass run twice.
// imagesFunc is the Engine.Images seam, named so the wrapper below reads.
type imagesFunc = func(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error)
// occupancy counts how many goroutines are inside a wrapped operation at once,
// per key, and remembers the high-water mark.
type occupancy struct {
mu sync.Mutex
in map[string]int
high map[string]int
}
func newOccupancy() *occupancy { return &occupancy{in: map[string]int{}, high: map[string]int{}} }
// watchImages wraps an Images func with the gauge, keyed by image sha — which
// is a VM's identity for a VM built with ownImage.
func (o *occupancy) watchImages(next imagesFunc) imagesFunc {
return func(ctx context.Context, url, sha string, progress func(done, total int64)) (string, error) {
o.mu.Lock()
o.in[sha]++
o.high[sha] = max(o.high[sha], o.in[sha])
o.mu.Unlock()
defer func() {
o.mu.Lock()
o.in[sha]--
o.mu.Unlock()
}()
return next(ctx, url, sha, progress)
}
}
func (o *occupancy) highWater(key string) int {
o.mu.Lock()
defer o.mu.Unlock()
return o.high[key]
}
// TestABusyVMNeverRunsTwoPassesAtOnce pins the layer's whole reason for
// existing: one goroutine per VM, so the worker IS the lock and a single VM's
// operations are serial without a per-VM mutex. A second pass loop for one VM
// puts a Destroy/DeleteVM alongside a Boot/SaveVM with nothing between them.
//
// The second assignment is what makes this observable: it lands while the first
// pass is parked, so one loop coalesces it into pending and nothing new enters,
// while a second loop consumes it immediately and walks into the provisioner
// beside the pass already there.
func TestABusyVMNeverRunsTwoPassesAtOnce(t *testing.T) {
f := setup(t)
occ := newOccupancy()
arrived := make(chan struct{}, 2)
release := make(chan struct{})
releaseAll := sync.OnceFunc(func() { close(release) })
t.Cleanup(releaseAll)
f.eng.Images = occ.watchImages(wedge(arrived, release))
stepNoWait(t, f, snap(1, vm("vm1", ownImage)))
select {
case <-arrived:
case <-time.After(2 * time.Second):
t.Fatal("worker never started the create")
}
stepNoWait(t, f, snap(2, vm("vm1", ownImage)))
select {
case <-arrived:
t.Fatal("two passes entered the provisioner for one VM at once: the worker is no longer the lock, so a Destroy can run against a Boot and orphan a hypervisor with no record left to find it by")
case <-time.After(250 * time.Millisecond):
}
releaseAll()
f.eng.manager().waitIdle()
assert.Equal(t, 1, occ.highWater("vm1"),
"a VM was inside its slow create more than once at a time: one goroutine per VM is what serializes this VM's operations")
}
// TestBusyWorkerIsNotReaped pins that reaping only ever takes an IDLE worker.
// A worker mid-pass owns its VM; dropping it from the map lets the next deliver
// for the same id spawn a SECOND worker, which is the two-goroutines-for-one-VM
// state this layer exists to prevent.
//
// reapAbsent is called directly with an empty live set because that is the
// production shape of the hazard: LoadVMs continues past read errors, so a
// transiently unreadable record drops a live VM out of live for one tick while
// its pass is still running. Driving it through Step cannot reproduce that —
// the record is readable, so the union puts the VM straight back.
func TestBusyWorkerIsNotReaped(t *testing.T) {
f := setup(t)
m := f.eng.manager()
arrived := make(chan struct{}, 1)
release := make(chan struct{})
releaseAll := sync.OnceFunc(func() { close(release) })
t.Cleanup(releaseAll)
f.eng.Images = wedge(arrived, release)
stepNoWait(t, f, snap(1, vm("vm1")))
select {
case <-arrived:
case <-time.After(2 * time.Second):
t.Fatal("worker never started the create")
}
m.reapAbsent(map[string]assignment{}) // the tick that could not read vm1's record
require.Equal(t, 1, m.count(),
"a busy worker was reaped: the next deliver for this VM spawns a second goroutine for it, and a concurrent Destroy against a Boot orphans a hypervisor process")
// The other half, so the pair reads as one rule: once the pass ends, the
// same absent VM IS reaped — deferring a busy worker costs only a tick.
releaseAll()
m.waitIdle()
m.reapAbsent(map[string]assignment{})
assert.Equal(t, 0, m.count(), "an idle worker for a VM that is gone must be reaped")
}
// TestEveryReportOwnsItsRows pins the clone in collect. A worker publishes once
// and then serves that result until its next pass ends, so without the clone
// every report taken in between carries the SAME row proto — and merge writes
// into the row it hands out (the level-triggered host key), so that write lands
// in reports other goroutines are already holding, and in the worker's own live
// result. Two Steps from concurrent sessions (an agent reconnect) is that case.
func TestEveryReportOwnsItsRows(t *testing.T) {
f := setup(t)
f.step(snap(1, vm("vm1", withHostCert("ssh-ed25519-cert-v01@openssh.com AAAA"))))
first := findVM(f.aggregateNow(1), "vm1")
second := findVM(f.aggregateNow(1), "vm1")
require.NotNil(t, first)
require.NotNil(t, second)
require.NotEmpty(t, first.GetSshHostPubkey(), "precondition: merge stamps the host key into the row it hands out")
assert.NotSame(t, first, second,
"two reports share one row proto: collect stopped cloning, so merge's write lands in every report already holding that row and in the worker's live result")
}