internal/agent/reconcile/narrate_test.go
Ref: Size: 8.8 KiB History
package reconcile
import (
"context"
"errors"
"sync"
"testing"
"github.com/a73x/eitri/internal/agent/state"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// narrateCreate runs one VM's create with a publisher attached and returns
// every sentence it published, in order. This is the seam the worker uses in
// production (see worker.publish) — a pass that has not finished still has
// something to say.
func narrateCreate(t *testing.T, f *fixture, id string) []string {
t.Helper()
var said []string
var res vmResult
f.eng.create(context.Background(), vm(id), state.Record{}, false, &res,
func(interim vmResult) {
require.Equal(t, "creating", interim.vm.GetPhase(), "narration never changes the phase")
said = append(said, interim.vm.GetStatusDetail())
})
return said
}
// TestACreateSaysWhatItIsDoing is the whole point of the field: `creating` is
// one word for image, disk, seed and boot, and only the host can see which of
// them a VM is sitting in.
func TestACreateSaysWhatItIsDoing(t *testing.T) {
f := setup(t)
said := narrateCreate(t, f, "vm1")
assert.Equal(t, []string{
"waiting for a create slot on this host",
"downloading image",
"preparing root disk",
"building the cloud-init seed",
"booting",
}, said)
}
// TestACreateCountsOutTheDownload pins the one step that takes long enough to
// need a number rather than a name.
func TestACreateCountsOutTheDownload(t *testing.T) {
f := setup(t)
f.eng.Images = func(_ context.Context, _, sha string, progress func(done, total int64)) (string, error) {
progress(1_288_490_188, 3_972_844_748) // 1.2 of 3.7 GiB
return "/cache/" + sha + ".raw", nil
}
assert.Contains(t, narrateCreate(t, f, "vm1"), "downloading image 1.2/3.7 GiB")
}
// TestNarrationCarriesTheGuestHostKey: the public host key is level-triggered —
// it must be in EVERY report for as long as the VM exists, or the control plane
// stops re-certifying it. A row published mid-pass is a report row like any
// other and cannot be the one that drops it.
func TestNarrationCarriesTheGuestHostKey(t *testing.T) {
f := setup(t)
var res vmResult
res.hostPubKey = "ssh-ed25519 AAAA-test guest"
published := 0
f.eng.create(context.Background(), vm("vm1"), state.Record{}, false, &res,
func(interim vmResult) {
published++
assert.Equal(t, res.hostPubKey, interim.hostPubKey)
})
assert.Positive(t, published)
}
// TestACreateWithNobodyListeningStillCreates: narration is commentary, and a
// pass with no publisher must behave exactly as it always did.
func TestACreateWithNobodyListeningStillCreates(t *testing.T) {
f := setup(t)
var res vmResult
f.eng.create(context.Background(), vm("vm1"), state.Record{}, false, &res, nil)
require.NotNil(t, res.vm)
assert.Equal(t, "ready", res.vm.GetPhase())
assert.Empty(t, res.vm.GetStatusDetail(), "a settled VM has nothing to add")
}
// TestNarrationCannotOutliveThePassThatSaidIt is the straggler case. The image
// cache shares one fetch between waiters, so its progress callback runs on a
// goroutine the pass does not own and keeps draining after the pass has given
// up on the download. A line landing after the pass published its failure would
// replace a settled "failed" row with "creating — downloading image", and the
// next host report would show a phantom in-flight create where a failure
// happened — narration would have become state.
//
// It also pins the feature the guard could quietly kill: while the pass is
// still live, its narration must land in the report as it always did.
func TestNarrationCannotOutliveThePassThatSaidIt(t *testing.T) {
f := setup(t)
f.eng.MaxCreateAttempts = 1 // one failure is terminal, so the final row is unambiguous
var mu sync.Mutex
var straggler func(done, total int64)
narrated := make(chan struct{})
release := make(chan struct{})
f.eng.Images = func(_ context.Context, _, _ string, progress func(done, total int64)) (string, error) {
mu.Lock()
straggler = progress // kept past the fetch, the way a shared fetch keeps it
mu.Unlock()
progress(1_288_490_188, 3_972_844_748)
close(narrated)
<-release
return "", errors.New("fetch gave up")
}
stepNoWait(t, f, snap(1, vm("vm1")))
<-narrated
row := findVM(f.aggregateNow(1), "vm1")
require.NotNil(t, row, "a live pass still narrates")
assert.Equal(t, "downloading image 1.2/3.7 GiB", row.GetStatusDetail())
close(release)
f.eng.manager().waitIdle()
require.Equal(t, "failed", findVM(f.aggregateNow(1), "vm1").GetPhase(), "precondition: the pass ended failed")
// The fetch goroutine drains on and says one more thing, after the pass that
// asked for it has published its failure.
mu.Lock()
say := straggler
mu.Unlock()
say(3_972_844_748, 3_972_844_748)
final := findVM(f.aggregateNow(1), "vm1")
require.NotNil(t, final)
assert.Equal(t, "failed", final.GetPhase(), "a straggler must not resurrect a finished create")
assert.Equal(t, "fetch gave up", final.GetLastError())
assert.Empty(t, final.GetStatusDetail())
}
// TestADeadPassSaysNothing: the source-side half. A pass whose context has
// expired stops narrating at all, so the goroutine still draining a shared
// fetch never calls into a pass that has given up on it.
func TestADeadPassSaysNothing(t *testing.T) {
f := setup(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
f.eng.Images = func(_ context.Context, _, sha string, progress func(done, total int64)) (string, error) {
cancel() // the watchdog fires while the fetch is still running
progress(1_288_490_188, 3_972_844_748)
return "/cache/" + sha + ".raw", nil
}
var said []string
var res vmResult
f.eng.create(ctx, vm("vm1"), state.Record{}, false, &res,
func(interim vmResult) { said = append(said, interim.vm.GetStatusDetail()) })
assert.NotContains(t, said, "downloading image 1.2/3.7 GiB")
}
func TestDownloadDetailWords(t *testing.T) {
assert.Equal(t, "downloading image 1.2/3.7 GiB", downloadDetail(1_288_490_188, 3_972_844_748))
// Both halves scale together off the larger, so the pair reads as a fraction
// rather than as arithmetic between two units.
assert.Equal(t, "downloading image 0.4/3.7 GiB", downloadDetail(429_496_730, 3_972_844_748))
assert.Equal(t, "downloading image 120.0/512.0 MiB", downloadDetail(125_829_120, 536_870_912))
// 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()
}