2596f764

test: the concurrency rules this package argues are now observed

a73x 2026-08-23 11:13

Commit message

test: the concurrency rules this package argues are now observed

Every exclusion invariant in the per-VM worker layer was stated in prose and
pinned by nothing: the suite asserts quiesced end-state, which is
interleaving-independent by construction, and -race is blind here by design —
every shared structure is individually mutex-guarded, so an execution running
two passes for one VM is race-clean.

What was missing was an observable read at the moment two passes differ from
one, plus the interleavings that force it:

  - occupancy, a per-key entry/exit gauge over the Images seam, composed with
    the existing wedge: a busy VM never runs two passes at once.
  - reapAbsent driven directly with an empty live set — the tick that could not
    read a record — proving a busy worker is not reaped.
  - a straggler from pass N invoked while pass N+1 is parked mid-fetch, the
    generation half of the publish guard that busy alone cannot express.
  - pointer non-identity between two reports, for the clone in collect.
  - park-inside-write for both trackers' write-through atomicity.

statusTracker's atomicity alarm is rebuilt on the same parked-writer pattern.
Its 300-pair loop detected an unlocked write on 3 of 12 single -race runs —
make test runs count=1 — and could not detect one at all except among the final
few pairs, because a divergence self-heals on the next write of the other
status.

internal/agent/reconcile/narrate_test.go

@@ @@ -165,3 +165,69 @@ func TestDownloadDetailWords(t *testing.T) { @@
+	// A server that declared no length gets the half it can vouch for.
+	assert.Equal(t, "downloading image 1.2 GiB", downloadDetail(1_288_490_188, -1))
+// TestAStragglerFromAnEarlierPassCannotLandOnALaterOne is the generation half
+// of the publish guard, and the sibling of the test above: that one kills a
+// straggler after its pass ENDED (busy is false), this one kills a straggler
+// from pass N while pass N+1 is in flight (busy is true, and only the
+// generation tells them apart). Without it the report shows a phantom state
+// from a pass that finished — the earlier pass's download progress on top of
+// the live pass's own row — and the control plane reads narration as state.
+//
+// Nothing races here: pass 2 is parked inside its fetch, so the straggler is
+// invoked from the test goroutine at a moment that is fully determined.
+func TestAStragglerFromAnEarlierPassCannotLandOnALaterOne(t *testing.T) {
+	f := setup(t)
+	var mu sync.Mutex
+	var straggler func(done, total int64)
+	passes := 0
+	firstFetched := make(chan struct{})
+	secondNarrated := make(chan struct{})
+	release := make(chan struct{})
+	releaseAll := sync.OnceFunc(func() { close(release) })
+	t.Cleanup(releaseAll)
+	f.eng.Images = func(_ context.Context, _, _ string, progress func(done, total int64)) (string, error) {
+		mu.Lock()
+		passes++
+		n := passes
+		if n == 1 {
+			straggler = progress // kept past the fetch, the way a shared fetch keeps it
+		}
+		mu.Unlock()
+		if n == 1 {
+			close(firstFetched)
+			return "", errors.New("fetch gave up")
+		}
+		progress(2_147_483_648, 4_294_967_296) // pass 2's own line: 2.0/4.0 GiB
+		close(secondNarrated)
+		<-release
+		return "", errors.New("fetch gave up")
+	}
+	stepNoWait(t, f, snap(1, vm("vm1")))
+	<-firstFetched
+	f.eng.manager().waitIdle() // pass 1 is over; its generation is closed
+	stepNoWait(t, f, snap(2, vm("vm1")))
+	<-secondNarrated
+	require.Equal(t, "downloading image 2.0/4.0 GiB",
+		findVM(f.aggregateNow(2), "vm1").GetStatusDetail(),
+		"precondition: pass 2 is in flight and narrating")
+	// Pass 1's fetch goroutine drains on and says one more thing — while pass 2
+	// is still running, so busy alone cannot tell it apart from pass 2's own.
+	mu.Lock()
+	say := straggler
+	mu.Unlock()
+	require.NotNil(t, say, "precondition: pass 1's progress callback outlived pass 1")
+	say(1_288_490_188, 3_972_844_748) // pass 1's line: 1.2/3.7 GiB
+	assert.Equal(t, "downloading image 2.0/4.0 GiB",
+		findVM(f.aggregateNow(2), "vm1").GetStatusDetail(),
+		"a line from pass 1 landed while pass 2 was in flight: the report now carries a phantom state from a finished pass, and the pass actually running is hidden behind it")
+	releaseAll()
+	f.eng.manager().waitIdle()
+}

internal/agent/reconcile/reconcile_test.go

Old		New
@@ -280,6 +280,14 @@ func vm(id string, opts ...func(pb.VMSpec)) pb.VMSpec {
280	func tombstoned(v pb.VMSpec) pb.VMSpec { v.Tombstoned = true; return v }	280	func tombstoned(v pb.VMSpec) pb.VMSpec { v.Tombstoned = true; return v }
281	func stopped(v *pb.VMSpec) { v.PowerState = "stopped" }	281	func stopped(v *pb.VMSpec) { v.PowerState = "stopped" }
282		282
		283	// ownImage gives a VM an image no other VM uses. The Images seam is told a URL
		284	// and a sha and never a vm_id, so this is what makes an observable hung on the
		285	// image cache a PER-VM observable — see occupancy.
		286	func ownImage(v *pb.VMSpec) {
		287	v.ImageUrl = "http://x/" + v.VmId + ".img"
		288	v.ImageSha256 = v.VmId
		289	}
		290
283	func findVM(rep pb.Report, id string) pb.VMStatus {	291	func findVM(rep pb.Report, id string) pb.VMStatus {
284	for _, v := range rep.Vms {	292	for _, v := range rep.Vms {
285	if v.VmId == id {	293	if v.VmId == id {

internal/agent/reconcile/worker_test.go

@@ @@ -2,6 +2,7 @@ package reconcile @@
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
@@ @@ -224,3 +225,143 @@ func TestCreateConcurrencyZeroIsUnlimited(t *testing.T) { @@
+	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")
+}

internal/server/syncsvc/tracker_test.go

Old		New
@@ -4,6 +4,7 @@ import (
4	"errors"	4	"errors"
5	"sync"	5	"sync"
6	"testing"	6	"testing"
		7	"time"
7		8
8	"github.com/stretchr/testify/require"	9	"github.com/stretchr/testify/require"
9	)	10	)
@@ -132,34 +133,84 @@ func TestStatusTrackerDroppedAddressNotCached(t *testing.T) {
132	}	133	}
133	}	134	}
134		135
135	// TestStatusTrackerWriteThroughAtomic pins the invariant Fix D restores: under	136	// parkedWriter is the shape both write-through atomicity tests need: a durable
136	// concurrent conflicting reports for one VM, the cached triple always equals the	137	// write that announces it has started and then holds still until released, so a
137	// LAST durable write. writeThrough holds the lock across write+commit, so the	138	// test can apply a second write for the same key at the one moment the two
138	// two can never be reordered across goroutines and leave the cache disagreeing	139	// could interleave. The lock either keeps the second writer out of its write —
139	// with the row (the divergence that would suppress every later report). Runs	140	// which is the invariant — or it does not, and the second writer says so
140	// under -race to exercise the interleavings.	141	// immediately. It is never a question of which goroutine wins a race: under a
		142	// broken lock the second writer is runnable and blocked on nothing.
		143	type parkedWriter struct {
		144	inside chan struct{} // closed once this writer is inside its durable write
		145	release chan struct{} // closed by the test to let it finish
		146	}
		147
		148	func newParkedWriter() *parkedWriter {
		149	return &parkedWriter{inside: make(chan struct{}), release: make(chan struct{})}
		150	}
		151
		152	// entered reports whether a writer got into its durable write within a window
		153	// long enough that only a starved goroutine could miss it.
		154	func entered(inside <-chan struct{}) bool {
		155	select {
		156	case <-inside:
		157	return true
		158	case <-time.After(250 * time.Millisecond):
		159	return false
		160	}
		161	}
		162
		163	// TestStatusTrackerWriteThroughAtomic pins that decide→write→commit runs under
		164	// one lock per VM. Concurrent reports for the same host — an agent reconnect,
		165	// where the old and new sessions both deliver a report — must not interleave
		166	// the durable write and the cache update, because a cache left disagreeing with
		167	// the row treats every later identical report as "unchanged" and suppresses it
		168	// forever.
		169	//
		170	// The exclusion is asserted directly rather than sampled: -race cannot see this
		171	// (the tracker's own mutex covers every shared access, so an unlocked write is
		172	// still race-clean), and an end-state check over many random pairs is both a
		173	// coin flip per run and self-healing — a divergence is papered over by the next
		174	// write of the other status, so only one among the final few survives to be
		175	// seen.
141	func TestStatusTrackerWriteThroughAtomic(t *testing.T) {	176	func TestStatusTrackerWriteThroughAtomic(t *testing.T) {
142	tr := newStatusTracker()	177	tr := newStatusTracker()
		178	first := newParkedWriter()
		179	secondInside := make(chan struct{})
143	var mu sync.Mutex	180	var mu sync.Mutex
144	lastWritten := "" // status of the most recent successful durable write	181	lastWritten := "" // status of the most recent successful durable write
145		182
146	writer := func(status string) {	183	go func() {
147	_ = tr.writeThrough("vm1", status, "", "10.0.0.1", func() (string, error) {	184	_ = tr.writeThrough("vm1", "ready", "", "10.0.0.1", func() (string, error) {
		185	close(first.inside)
		186	<-first.release
148	mu.Lock()	187	mu.Lock()
149	lastWritten = status // recorded inside writeThrough's critical section	188	lastWritten = "ready"
150	mu.Unlock()	189	mu.Unlock()
151	return "10.0.0.1", nil	190	return "10.0.0.1", nil
152	})	191	})
153	}	192	}()
		193	<-first.inside
154		194
155	var wg sync.WaitGroup	195	done := make(chan struct{})
156	for range 300 {	196	go func() {
157	wg.Add(2)	197	defer close(done)
158	go func() { defer wg.Done(); writer("ready") }()	198	_ = tr.writeThrough("vm1", "failed", "", "10.0.0.1", func() (string, error) {
159	go func() { defer wg.Done(); writer("failed") }()	199	close(secondInside)
		200	mu.Lock()
		201	lastWritten = "failed"
		202	mu.Unlock()
		203	return "10.0.0.1", nil
		204	})
		205	}()
		206
		207	if entered(secondInside) {
		208	close(first.release)
		209	t.Fatal("a second report for the same VM entered its durable write while the first was mid-write: decide→write→commit is no longer atomic, so the cache can commit a status the row does not hold and suppress every later report of the true one")
160	}	210	}
161	wg.Wait()
162		211
		212	close(first.release)
		213	<-done
163	tr.mu.Lock()	214	tr.mu.Lock()
164	cached := tr.last["vm1"].status	215	cached := tr.last["vm1"].status
165	tr.mu.Unlock()	216	tr.mu.Unlock()
@@ -169,6 +220,58 @@ func TestStatusTrackerWriteThroughAtomic(t *testing.T) {
169	require.Equal(t, dbLast, cached, "cache must equal the last durable write")	220	require.Equal(t, dbLast, cached, "cache must equal the last durable write")
170	}	221	}
171		222
		223	// TestNetTrackerWriteThroughAtomic is the same invariant for the per-host facts
		224	// (subnet, host address). Two connections for one host — the reconnect again —
		225	// must not interleave: a cache holding a value the row never took makes every
		226	// later report of the true value read as "unchanged", and the fleet keeps
		227	// serving the stale subnet for as long as the process lives.
		228	func TestNetTrackerWriteThroughAtomic(t *testing.T) {
		229	tr := newNetTracker()
		230	first := newParkedWriter()
		231	secondInside := make(chan struct{})
		232	var mu sync.Mutex
		233	lastWritten := ""
		234
		235	go func() {
		236	_ = tr.writeThrough("h1", "10.77.1.0/24", func() error {
		237	close(first.inside)
		238	<-first.release
		239	mu.Lock()
		240	lastWritten = "10.77.1.0/24"
		241	mu.Unlock()
		242	return nil
		243	})
		244	}()
		245	<-first.inside
		246
		247	done := make(chan struct{})
		248	go func() {
		249	defer close(done)
		250	_ = tr.writeThrough("h1", "192.168.64.0/24", func() error {
		251	close(secondInside)
		252	mu.Lock()
		253	lastWritten = "192.168.64.0/24"
		254	mu.Unlock()
		255	return nil
		256	})
		257	}()
		258
		259	if entered(secondInside) {
		260	close(first.release)
		261	t.Fatal("a second write for the same host entered while the first was mid-write: decide→write→commit is no longer atomic per host, so the cache can hold a subnet the row never took and suppress every later report of the real one")
		262	}
		263
		264	close(first.release)
		265	<-done
		266	tr.mu.Lock()
		267	cached := tr.last["h1"]
		268	tr.mu.Unlock()
		269	mu.Lock()
		270	dbLast := lastWritten
		271	mu.Unlock()
		272	require.Equal(t, dbLast, cached, "cache must equal the last durable write")
		273	}
		274
172	func TestStatusTrackerForget(t *testing.T) {	275	func TestStatusTrackerForget(t *testing.T) {
173	tr := newStatusTracker()	276	tr := newStatusTracker()
174	if !tr.write("vm1", "ready", "", "10.0.0.1") {	277	if !tr.write("vm1", "ready", "", "10.0.0.1") {