internal/server/syncsvc/syncsvc_test.go
Ref: Size: 44.9 KiB History
package syncsvc
import (
"context"
"database/sql"
"errors"
"io"
"runtime"
"strings"
"sync"
"testing"
"time"
"github.com/a73x/eitri/internal/pb"
"github.com/a73x/eitri/internal/server/hosttoken"
"github.com/a73x/eitri/internal/server/hub"
"github.com/a73x/eitri/internal/server/registry"
"github.com/a73x/eitri/internal/server/store"
"github.com/a73x/eitri/internal/transport"
"github.com/quic-go/quic-go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
type fixture struct {
st *store.Store
reg *registry.Registry
hub *hub.Hub
addr string
fp string
secret []byte
host store.Host
cred string
svc *Service
}
func setup(t *testing.T) *fixture {
t.Helper()
return setupWithWriteTimeout(t, 0) // 0 → production default
}
// testTenant is the tenant the test builders provision — through the real JIT
// path, the only way tenants are born. The name has no significance.
const testTenant = "default"
// seedTestTenant JIT-provisions testTenant on a fresh store.
func seedTestTenant(t *testing.T, st *store.Store) {
t.Helper()
_, err := st.CreateTenantForIdentity("https://test-issuer", "test-subject", testTenant+"@test.local")
require.NoError(t, err)
}
func setupWithWriteTimeout(t *testing.T, writeTimeout time.Duration) *fixture {
t.Helper()
st, err := store.Open(t.TempDir()+"/db", "10.77.0.0/16")
require.NoError(t, err)
t.Cleanup(func() { st.Close() })
seedTestTenant(t, st)
tok, _ := st.CreateEnrollmentToken(testTenant)
host, err := st.RedeemEnrollmentToken(tok, store.EnrollFacts{Name: "h", OS: "linux", Arch: "amd64", Provisioner: "cloudhv", Remote: ""})
require.NoError(t, err)
reg := registry.New(time.Now)
h := hub.New()
secret := []byte("s3cret")
addr, fp, svc, _ := startTestServer(t, st, reg, h, secret, writeTimeout)
return &fixture{st: st, reg: reg, hub: h, addr: addr, fp: fp, secret: secret,
host: host, cred: hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now()), svc: svc}
}
// startTestServer listens on 127.0.0.1:0 (random UDP port) and returns the addr,
// the server cert fingerprint, the service, and a cleanup func. A zero
// writeTimeout uses the production default.
func startTestServer(t *testing.T, st *store.Store, reg *registry.Registry, h *hub.Hub, secret []byte, writeTimeout time.Duration) (addr, fp string, svc *Service, stop func()) {
t.Helper()
certPEM, keyPEM, err := transport.GenerateServerCert()
require.NoError(t, err)
fp, err = transport.CertFingerprint(certPEM)
require.NoError(t, err)
tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
require.NoError(t, err)
lis, err := quic.ListenAddr("127.0.0.1:0", tlsConf, &quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
svc = newWithWriteTimeout(st, reg, h, secret, 0, writeTimeout)
ctx, cancel := context.WithCancel(context.Background())
go svc.Serve(ctx, lis) //nolint:errcheck
stop = func() { cancel(); lis.Close() }
t.Cleanup(stop)
return lis.Addr().String(), fp, svc, stop
}
// conn is the agent-side dual-stream connection used by the test scenarios.
type testConn struct {
conn quic.Connection
up quic.Stream
down quic.Stream
}
func (c *testConn) send(t *testing.T, msg *pb.AgentMessage) {
t.Helper()
require.NoError(t, transport.WriteMsg(c.up, msg))
}
func (c *testConn) recv(t *testing.T) *pb.ServerMessage {
t.Helper()
var msg pb.ServerMessage
require.NoError(t, transport.ReadMsg(c.down, &msg, transport.DefaultMaxFrame))
return &msg
}
// dial opens a QUIC connection pinned to fp, opens the up-stream, sends the
// Hello (carrying cred), and accepts the server's down-stream. Returns the conn
// or an error from AcceptStream (auth rejection surfaces there).
func dial(t *testing.T, addr, fp, hostID, cred string) (*testConn, error) {
t.Helper()
return dialAs(t, addr, fp, hostID, cred, "cloudhv")
}
// dialAs is dial with the provisioner the Hello claims, so a test can watch a
// host change backend.
func dialAs(t *testing.T, addr, fp, hostID, cred, provisioner string) (*testConn, error) {
t.Helper()
return dialHello(t, addr, fp, &pb.Hello{HostId: hostID, Provisioner: provisioner, Credential: cred})
}
// dialHello is the connect handshake with the Hello spelled out, for the tests
// that care what an agent says about itself on the way in.
func dialHello(t *testing.T, addr, fp string, h *pb.Hello) (*testConn, error) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
conn, err := quic.DialAddr(ctx, addr, transport.ClientTLS(fp),
&quic.Config{MaxIdleTimeout: 5 * time.Second})
if err != nil {
return nil, err
}
up, err := conn.OpenStreamSync(ctx)
if err != nil {
return nil, err
}
hello := &pb.AgentMessage{Msg: &pb.AgentMessage_Hello{Hello: h}}
if err := transport.WriteMsg(up, hello); err != nil {
return nil, err
}
down, err := conn.AcceptStream(ctx)
if err != nil {
return nil, err
}
c := &testConn{conn: conn, up: up, down: down}
t.Cleanup(func() { conn.CloseWithError(0, "") })
return c, nil
}
func mustDial(t *testing.T, f *fixture) *testConn {
t.Helper()
c, err := dial(t, f.addr, f.fp, f.host.ID, f.cred)
require.NoError(t, err)
return c
}
func TestRejectsBadCredential(t *testing.T) {
f := setup(t)
// A junk credential: the server CloseWithError(CodeAuthRejected) surfaces on
// the agent's AcceptStream as a *quic.ApplicationError.
_, err := dial(t, f.addr, f.fp, f.host.ID, "host-x.deadbeef")
require.Error(t, err, "stream must be terminated")
var appErr *quic.ApplicationError
require.True(t, errors.As(err, &appErr), "expected *quic.ApplicationError, got %T: %v", err, err)
assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
}
func TestPushesSnapshotOnConnectAndOnPoke(t *testing.T) {
f := setup(t)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
c := mustDial(t, f)
msg := c.recv(t)
snap := msg.GetSnapshot()
require.NotNil(t, snap)
require.Len(t, snap.Vms, 1)
assert.Equal(t, "vm1", snap.Vms[0].VmId)
first := snap.Epoch
// A desired-state edit + poke re-pushes with a higher epoch.
require.NoError(t, f.st.SetVMPower("vm1", "stopped"))
f.hub.Poke(f.host.ID)
msg = c.recv(t)
snap = msg.GetSnapshot()
assert.Greater(t, snap.Epoch, first)
assert.Equal(t, "stopped", snap.Vms[0].PowerState)
}
func TestReportWritesThroughAndHardDeletesAckedTombstones(t *testing.T) {
f := setup(t)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
require.NoError(t, f.st.TombstoneVM("vm1"))
c := mustDial(t, f)
c.recv(t) // initial snapshot
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{
Destroyed: []string{"vm1"}, // level-triggered ack
Capacity: &pb.Capacity{Vcpus: 8},
LastSeenEpoch: 2,
}}})
require.Eventually(t, func() bool {
vms, _ := f.st.ListVMs()
return len(vms) == 0
}, 2*time.Second, 20*time.Millisecond, "acked tombstone must be hard-deleted")
st, ok := f.reg.Get(f.host.ID)
require.True(t, ok)
assert.Equal(t, int64(8), st.Capacity.VCPUs)
}
// countingRecorder wraps a real recorder and counts (with a mutex, since host
// read-loop goroutines call concurrently) how many RecordVMStatus writes reach
// the store, so a test can prove that a repeated unchanged report performs none.
type countingRecorder struct {
mu sync.Mutex
inner vmStatusRecorder
calls int
}
func (c *countingRecorder) RecordVMStatus(id, hostID, status, lastErr, ip string) (string, error) {
c.mu.Lock()
c.calls++
c.mu.Unlock()
return c.inner.RecordVMStatus(id, hostID, status, lastErr, ip)
}
func (c *countingRecorder) count() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.calls
}
// TestApplyReportSkipsUnchangedWrites proves the dedup: the SAME ready report
// sent twice performs exactly ONE durable write; a subsequent report that
// changes the ip performs another; and the DB's final (status, assigned_ip) is
// correct across the unchanged→changed sequence.
func TestApplyReportSkipsUnchangedWrites(t *testing.T) {
f := setup(t)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
// Swap in a counting recorder before any report is applied.
rec := &countingRecorder{inner: f.svc.recorder}
f.svc.recorder = rec
c := mustDial(t, f)
c.recv(t) // initial snapshot
ready := func(ip string) *pb.AgentMessage {
return &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{Vms: []*pb.VMStatus{
{VmId: "vm1", PowerState: "running", Phase: "ready", Ip: ip},
}}}}
}
// First report writes and lands in the DB.
c.send(t, ready("10.77.1.2"))
require.Eventually(t, func() bool {
vms, _ := f.st.ListVMs()
return len(vms) == 1 && vms[0].Status == "ready" && vms[0].AssignedIP == "10.77.1.2"
}, 2*time.Second, 20*time.Millisecond, "first report must write through")
require.Equal(t, 1, rec.count(), "first report writes exactly once")
// Identical report: must perform NO additional write. Assert the count stays
// at 1 for a sustained window (a redundant write would bump it).
c.send(t, ready("10.77.1.2"))
require.Never(t, func() bool {
return rec.count() != 1
}, 500*time.Millisecond, 25*time.Millisecond, "unchanged report must not write again")
// A changed ip must write again and update the DB.
c.send(t, ready("10.77.1.3"))
require.Eventually(t, func() bool {
return rec.count() == 2
}, 2*time.Second, 20*time.Millisecond, "changed ip must write again")
require.Eventually(t, func() bool {
vms, _ := f.st.ListVMs()
return len(vms) == 1 && vms[0].Status == "ready" && vms[0].AssignedIP == "10.77.1.3"
}, 2*time.Second, 20*time.Millisecond, "changed ip must land in the DB")
}
func TestReportWithValidIPUpdatesRegistryAndStore(t *testing.T) {
f := setup(t)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
c := mustDial(t, f)
c.recv(t)
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{Vms: []*pb.VMStatus{
{VmId: "vm1", PowerState: "running", Phase: "ready", Ip: "10.77.1.2"},
}}}})
require.Eventually(t, func() bool {
vms, _ := f.st.ListVMs()
return len(vms) == 1 && vms[0].Status == "ready" && vms[0].AssignedIP == "10.77.1.2"
}, 2*time.Second, 20*time.Millisecond)
}
// TestReportRecoversFromAnUnusableAddress walks the whole path a guest takes
// when its first answer is a bad one: an address the store refuses must not
// poison the write path against the address that follows it. A host whose guests
// live outside anything the fleet allocated — a Mac's vmnet, say — reports one
// of those on every tick, so "the second report is treated as unchanged" means
// blank forever, not blank once.
func TestReportRecoversFromAnUnusableAddress(t *testing.T) {
f := setup(t)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
c := mustDial(t, f)
c.recv(t)
ready := func(ip string) *pb.AgentMessage {
return &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{Vms: []*pb.VMStatus{
{VmId: "vm1", PowerState: "running", Phase: "ready", Ip: ip},
}}}}
}
// DHCP never answered: the guest reports the address it made up for itself.
c.send(t, ready("169.254.11.2"))
require.Eventually(t, func() bool {
vms, _ := f.st.ListVMs()
return len(vms) == 1 && vms[0].Status == "ready"
}, 2*time.Second, 20*time.Millisecond, "the status must write even so")
vms, _ := f.st.ListVMs()
require.Empty(t, vms[0].AssignedIP, "an unusable address must not land")
// DHCP answers, on a subnet no fleet allocation contains.
c.send(t, ready("192.168.64.7"))
require.Eventually(t, func() bool {
vms, _ := f.st.ListVMs()
return len(vms) == 1 && vms[0].AssignedIP == "192.168.64.7"
}, 2*time.Second, 20*time.Millisecond, "the address that works must land")
}
func TestHardDeleteTriggersRepush(t *testing.T) {
f := setup(t)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
require.NoError(t, f.st.TombstoneVM("vm1"))
c := mustDial(t, f)
// Receive initial snapshot (contains vm1 tombstoned).
snap := c.recv(t).GetSnapshot()
require.NotNil(t, snap)
require.Len(t, snap.Vms, 1)
// Send a report acking the destroyed VM.
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{
Destroyed: []string{"vm1"},
LastSeenEpoch: snap.Epoch,
}}})
// Expect a second snapshot push after the hard-delete — Vms list must be empty.
snap2 := c.recv(t).GetSnapshot()
require.NotNil(t, snap2)
assert.Empty(t, snap2.Vms, "snapshot after hard-delete must have no VMs")
}
func TestNoGoroutineLeakOnDisconnect(t *testing.T) {
f := setup(t)
// Warm to steady state: open several connections, read a snapshot, close.
const warm = 5
for range warm {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
conn, err := quic.DialAddr(ctx, f.addr, transport.ClientTLS(f.fp),
&quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
up, err := conn.OpenStreamSync(ctx)
require.NoError(t, err)
require.NoError(t, transport.WriteMsg(up, &pb.AgentMessage{Msg: &pb.AgentMessage_Hello{
Hello: &pb.Hello{HostId: f.host.ID, Credential: f.cred}}}))
down, err := conn.AcceptStream(ctx)
require.NoError(t, err)
var m pb.ServerMessage
require.NoError(t, transport.ReadMsg(down, &m, transport.DefaultMaxFrame))
conn.CloseWithError(0, "")
cancel()
}
time.Sleep(500 * time.Millisecond) // let server-side handlers tear down
before := runtime.NumGoroutine()
const n = 5
for range n {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
conn, err := quic.DialAddr(ctx, f.addr, transport.ClientTLS(f.fp),
&quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
up, err := conn.OpenStreamSync(ctx)
require.NoError(t, err)
require.NoError(t, transport.WriteMsg(up, &pb.AgentMessage{Msg: &pb.AgentMessage_Hello{
Hello: &pb.Hello{HostId: f.host.ID, Credential: f.cred}}}))
down, err := conn.AcceptStream(ctx)
require.NoError(t, err)
var m pb.ServerMessage
require.NoError(t, transport.ReadMsg(down, &m, transport.DefaultMaxFrame))
conn.CloseWithError(0, "")
cancel()
}
// Each disconnected connection must tear down its handler (read loop + poke
// goroutine). Allow small headroom for quic-go's own per-connection cleanup;
// a poke-goroutine leak would grow by one per connection beyond that.
require.Eventually(t, func() bool {
return runtime.NumGoroutine() <= before+n+2
}, 5*time.Second, 50*time.Millisecond, "handler goroutines must exit on disconnect")
}
// TestStalledReaderDoesNotPinServer asserts that an authenticated agent which
// keeps its QUIC connection alive but never reads the down-stream cannot pin the
// server's poke/read goroutines forever. With a short write timeout the server
// must close the connection once the flow-control window fills and a down-stream
// write hits the deadline.
//
// Against the unbounded-write code this test hangs: the poke goroutine blocks in
// Write (window full), cancel() does not unblock it, and the read loop is parked
// in ReadMsg(up) — so conn.Context() never fires and this asserts to failure.
func TestStalledReaderDoesNotPinServer(t *testing.T) {
f := setupWithWriteTimeout(t, 300*time.Millisecond)
// A VM with a large CloudInit blob so a few snapshots fill the flow-control
// window of the unread down-stream quickly.
bigBlob := strings.Repeat("x", 256*1024)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5,
PowerState: "running", CloudInit: bigBlob}))
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
conn, err := quic.DialAddr(ctx, f.addr, transport.ClientTLS(f.fp),
&quic.Config{KeepAlivePeriod: 15 * time.Second, MaxIdleTimeout: 30 * time.Second})
require.NoError(t, err)
defer conn.CloseWithError(0, "")
up, err := conn.OpenStreamSync(ctx)
require.NoError(t, err)
require.NoError(t, transport.WriteMsg(up, &pb.AgentMessage{Msg: &pb.AgentMessage_Hello{
Hello: &pb.Hello{HostId: f.host.ID, Credential: f.cred}}}))
// Deliberately DO NOT AcceptStream/read the down-stream. Poke repeatedly so
// the server keeps trying to write snapshots until the window fills and the
// next write hits the 300ms deadline.
go func() {
for range 100 {
f.hub.Poke(f.host.ID)
time.Sleep(20 * time.Millisecond)
}
}()
// The server must close the connection promptly once a write times out. This
// is deterministic: it is driven by the injected 300ms write timeout, not by
// any wall-clock guess about how long writes "should" take.
select {
case <-conn.Context().Done():
// pass: server closed the connection, unblocking its goroutines.
case <-time.After(5 * time.Second):
t.Fatal("server did not close the stalled connection — poke/read goroutines are pinned")
}
}
// TestStalledPreAuthConnDropped pins the pre-auth DoS guard: a peer that
// completes the QUIC handshake and opens the stream but never sends a full
// Hello must be dropped by the hello grace, not parked forever holding a
// goroutine. Writing a single byte surfaces the stream to the server's
// AcceptStream yet can never form a Hello frame (the 4-byte length prefix alone
// is incomplete), so the guard must fire on the Hello read deadline.
func TestStalledPreAuthConnDropped(t *testing.T) {
orig := helloGrace
helloGrace = 150 * time.Millisecond
t.Cleanup(func() { helloGrace = orig })
f := setup(t)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
conn, err := quic.DialAddr(ctx, f.addr, transport.ClientTLS(f.fp),
&quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
stream, err := conn.OpenStreamSync(ctx)
require.NoError(t, err)
_, err = stream.Write([]byte{0})
require.NoError(t, err)
start := time.Now()
_ = stream.SetReadDeadline(time.Now().Add(5 * time.Second))
buf := make([]byte, 1)
_, err = stream.Read(buf)
require.Error(t, err)
var appErr *quic.ApplicationError
require.True(t, errors.As(err, &appErr), "server must close the conn, got %T: %v", err, err)
assert.Less(t, time.Since(start), 3*time.Second, "stalled conn must drop near the hello grace")
}
func TestWrongCertPinRejected(t *testing.T) {
f := setup(t)
wrongFP := strings.Repeat("00", 32)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
conn, err := quic.DialAddr(ctx, f.addr, transport.ClientTLS(wrongFP),
&quic.Config{MaxIdleTimeout: 5 * time.Second})
if err == nil {
// Handshake may complete lazily; the failure surfaces on first stream op.
_, err = conn.OpenStreamSync(ctx)
}
require.Error(t, err, "wrong cert pin must fail the TLS handshake")
assert.Contains(t, strings.ToLower(err.Error()), "fingerprint",
"expected cert fingerprint mismatch, got: %v", err)
}
// TestStaleGenerationCredentialRejected pins per-host revocation: after
// BumpCredGeneration, a credential minted at the old generation is rejected
// with CodeAuthRejected, while a freshly-minted one connects.
func TestStaleGenerationCredentialRejected(t *testing.T) {
f := setup(t)
oldCred := hosttoken.Mint(f.secret, f.host.ID, f.host.CredGeneration, time.Now())
_, err := f.st.BumpCredGeneration(f.host.ID, "")
require.NoError(t, err)
_, err = dial(t, f.addr, f.fp, f.host.ID, oldCred)
require.Error(t, err, "revoked-generation credential must be rejected")
var appErr *quic.ApplicationError
require.ErrorAs(t, err, &appErr)
assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
fresh := hosttoken.Mint(f.secret, f.host.ID, f.host.CredGeneration+1, time.Now())
c, err := dial(t, f.addr, f.fp, f.host.ID, fresh)
require.NoError(t, err, "current-generation credential must connect")
c.conn.CloseWithError(0, "")
}
// TestExpiredCredentialRejectedWhenMaxAgeSet pins the optional max-age
// policy: with maxAge configured, a credential older than maxAge is rejected;
// with maxAge zero (default) age is ignored.
func TestExpiredCredentialRejectedWhenMaxAgeSet(t *testing.T) {
st, err := store.Open(t.TempDir()+"/db", "10.77.0.0/16")
require.NoError(t, err)
t.Cleanup(func() { st.Close() })
seedTestTenant(t, st)
tok, _ := st.CreateEnrollmentToken(testTenant)
host, err := st.RedeemEnrollmentToken(tok, store.EnrollFacts{Name: "h", OS: "linux", Arch: "amd64", Provisioner: "cloudhv", Remote: ""})
require.NoError(t, err)
reg := registry.New(time.Now)
h := hub.New()
secret := []byte("s3cret")
certPEM, keyPEM, err := transport.GenerateServerCert()
require.NoError(t, err)
fp, err := transport.CertFingerprint(certPEM)
require.NoError(t, err)
tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
require.NoError(t, err)
lis, err := quic.ListenAddr("127.0.0.1:0", tlsConf, &quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
svc := New(st, reg, h, secret, 30*24*time.Hour) // maxAge 30d
ctx, cancel := context.WithCancel(context.Background())
go svc.Serve(ctx, lis) //nolint:errcheck
t.Cleanup(func() { cancel(); lis.Close() })
addr := lis.Addr().String()
stale := hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now().Add(-31*24*time.Hour))
_, err = dial(t, addr, fp, host.ID, stale)
require.Error(t, err, "over-max-age credential must be rejected")
var appErr *quic.ApplicationError
require.ErrorAs(t, err, &appErr)
assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
fresh := hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now())
c, err := dial(t, addr, fp, host.ID, fresh)
require.NoError(t, err, "fresh credential must connect")
c.conn.CloseWithError(0, "")
}
// TestMaxAgeEnforcedMidSession pins that a live session does not outlive
// credential_max_age: the per-report re-check closes it once the credential's
// issued-at falls out of the window.
func TestMaxAgeEnforcedMidSession(t *testing.T) {
st, err := store.Open(t.TempDir()+"/db", "10.77.0.0/16")
require.NoError(t, err)
t.Cleanup(func() { st.Close() })
seedTestTenant(t, st)
tok, _ := st.CreateEnrollmentToken(testTenant)
host, err := st.RedeemEnrollmentToken(tok, store.EnrollFacts{Name: "h", OS: "linux", Arch: "amd64", Provisioner: "cloudhv", Remote: ""})
require.NoError(t, err)
reg := registry.New(time.Now)
h := hub.New()
secret := []byte("s3cret")
certPEM, keyPEM, err := transport.GenerateServerCert()
require.NoError(t, err)
fp, err := transport.CertFingerprint(certPEM)
require.NoError(t, err)
tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
require.NoError(t, err)
lis, err := quic.ListenAddr("127.0.0.1:0", tlsConf, &quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
// Timing is truncation-aware: Mint stores issued_unix at SECOND
// granularity, so the observed age can read up to 0.999s older than
// real. maxAge 3s with a 1.5s-old credential leaves >1s margin on the
// Hello side (worst case 2.499s < 3s), and the 1.8s sleep pushes the
// real age to 3.3s — past maxAge even before truncation inflation.
svc := New(st, reg, h, secret, 3*time.Second)
ctx, cancel := context.WithCancel(context.Background())
go svc.Serve(ctx, lis) //nolint:errcheck
t.Cleanup(func() { cancel(); lis.Close() })
cred := hosttoken.Mint(secret, host.ID, host.CredGeneration, time.Now().Add(-1500*time.Millisecond))
c, err := dial(t, lis.Addr().String(), fp, host.ID, cred)
require.NoError(t, err, "not-yet-expired credential connects")
defer c.conn.CloseWithError(0, "")
time.Sleep(1800 * time.Millisecond) // credential ages past maxCredAge
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{Report: &pb.Report{}}})
// The server must close the connection with CodeAuthRejected; the next
// read on the down-stream surfaces it.
var msg pb.ServerMessage
readErr := transport.ReadMsg(c.down, &msg, transport.DefaultMaxFrame)
if readErr == nil {
// First read may deliver the initial snapshot; the close lands next.
readErr = transport.ReadMsg(c.down, &msg, transport.DefaultMaxFrame)
}
require.Error(t, readErr)
var appErr *quic.ApplicationError
require.ErrorAs(t, readErr, &appErr)
assert.Equal(t, quic.ApplicationErrorCode(transport.CodeAuthRejected), appErr.ErrorCode)
}
// TestReconnectKeepsNewConnRegistered pins the deregistration guard in
// handleConn: when a host reconnects, the NEW connection's registration
// overwrites the old one's, and the OLD connection's deferred deregistration
// must not delete the new entry — OpenConsole must still reach the host.
func TestReconnectKeepsNewConnRegistered(t *testing.T) {
f := setup(t)
a := mustDial(t, f)
a.recv(t) // initial snapshot: conn A fully registered
b := mustDial(t, f)
b.recv(t) // conn B registered for the same host, overwriting A's entry
// B answers every console handshake with ok=true so OpenConsole completes.
go func() {
for {
cs, err := b.conn.AcceptStream(context.Background())
if err != nil {
return
}
var open pb.ServerMessage
if transport.ReadMsg(cs, &open, transport.DefaultMaxFrame) != nil {
return
}
_ = transport.WriteMsg(cs, &pb.AgentMessage{Msg: &pb.AgentMessage_ConsoleOpened{
ConsoleOpened: &pb.ConsoleOpened{Ok: true}}})
}
}()
// A has already been evicted by B's arrival; closing it from this end too
// changes nothing and is what a real displaced agent does next. Either way
// A's handler deregisters asynchronously, and the guard (only delete if the
// map still holds OUR conn) must leave B alone — so across the whole
// teardown window no attempt may ever see ErrAgentOffline.
a.conn.CloseWithError(0, "")
require.Never(t, func() bool {
stream, err := f.svc.OpenConsole(context.Background(), f.host.ID, "vm-1")
if err == nil {
stream.Close()
}
return errors.Is(err, ErrAgentOffline)
}, time.Second, 100*time.Millisecond,
"old conn's deregistration deleted the new conn's registry entry")
// And the console must actually still open on the surviving connection.
stream, err := f.svc.OpenConsole(context.Background(), f.host.ID, "vm-1")
require.NoError(t, err, "OpenConsole must keep working on the surviving connection")
stream.Close()
}
func TestOpenConsoleNoAgent(t *testing.T) {
svc := New(nil, nil, nil, []byte("s"), 0) // no store use on this path
_, err := svc.OpenConsole(context.Background(), "host-x", "vm-1")
assert.ErrorIs(t, err, ErrAgentOffline)
}
func TestOpenTCPNoAgent(t *testing.T) {
svc := New(nil, nil, nil, []byte("s"), 0) // no store use on this path
_, err := svc.OpenTCP(context.Background(), "host-x", "vm-1", 22)
assert.ErrorIs(t, err, ErrAgentOffline)
}
// TestOpenTCPBridgesBytes drives the full TCPOpen/TCPOpened handshake against a
// fake agent that answers ok=true, then asserts a usable RWC is returned and
// bytes pass both directions.
//
// The round trip happens after a pause LONGER than the injected handshake
// timeout, which is what makes it evidence that the deadline was cleared: a
// residual deadline would kill every console and every tunnel a fixed time
// after it opened, and sessions here are long-lived by definition.
func TestOpenTCPBridgesBytes(t *testing.T) {
f := setup(t)
f.svc.handshakeTimeout = 200 * time.Millisecond
c := mustDial(t, f)
c.recv(t) // initial snapshot
// Fake agent: accept the tunnel stream, read TCPOpen, reply ok=true, then
// echo whatever the server writes so we can prove bytes bridge.
gotPort := make(chan uint32, 1)
go func() {
st, err := c.conn.AcceptStream(context.Background())
if err != nil {
return
}
var open pb.ServerMessage
if transport.ReadMsg(st, &open, transport.DefaultMaxFrame) != nil {
return
}
gotPort <- open.GetTcpOpen().GetPort()
if transport.WriteMsg(st, &pb.AgentMessage{Msg: &pb.AgentMessage_TcpOpened{
TcpOpened: &pb.TCPOpened{Ok: true}}}) != nil {
return
}
// Echo raw bytes back after the handshake.
buf := make([]byte, 32)
for {
n, err := st.Read(buf)
if n > 0 {
_, _ = st.Write(buf[:n])
}
if err != nil {
return
}
}
}()
rwc, err := f.svc.OpenTCP(context.Background(), f.host.ID, "vm-1", 8080)
require.NoError(t, err)
defer rwc.Close()
assert.Equal(t, uint32(8080), <-gotPort, "agent must receive the requested port")
// Outlive the handshake deadline, then round-trip.
time.Sleep(300 * time.Millisecond)
_, err = rwc.Write([]byte("ping"))
require.NoError(t, err, "the handshake deadline must be cleared: a session that dies once it elapses is a console that drops mid-use")
buf := make([]byte, 4)
_, err = io.ReadFull(rwc, buf)
require.NoError(t, err, "the handshake deadline must be cleared: a session that dies once it elapses is a console that drops mid-use")
assert.Equal(t, "ping", string(buf))
}
// TestOpenTCPAbandonsAWedgedAgent pins the other half of the deadline's life:
// an agent that accepts the stream and never answers must not hold the caller.
// Without the deadline the WS handler goroutine waits forever on a read no one
// will satisfy — a silent leak, one goroutine per attempted console.
func TestOpenTCPAbandonsAWedgedAgent(t *testing.T) {
f := setup(t)
f.svc.handshakeTimeout = 100 * time.Millisecond
c := mustDial(t, f)
c.recv(t) // initial snapshot
accepted := make(chan struct{})
go func() {
st, err := c.conn.AcceptStream(context.Background())
if err != nil {
return
}
var open pb.ServerMessage
if transport.ReadMsg(st, &open, transport.DefaultMaxFrame) != nil {
return
}
close(accepted) // the open arrived; the reply never will
<-time.After(10 * time.Second)
}()
done := make(chan error, 1)
go func() {
_, err := f.svc.OpenTCP(context.Background(), f.host.ID, "vm-1", 22)
done <- err
}()
<-accepted
select {
case err := <-done:
require.Error(t, err, "a silent agent must fail the open, not return a stream nobody is on the far end of")
assert.Contains(t, err.Error(), "tcp reply", "the failure must name the read that timed out")
case <-time.After(3 * time.Second):
t.Fatal("OpenTCP is still waiting on a wedged agent: without the handshake deadline this goroutine never returns, and one leaks per attempted session")
}
}
// TestOpenTCPRefused pins the ok=false path: the agent replies with an error
// string and OpenTCP surfaces it (not a usable stream).
func TestOpenTCPRefused(t *testing.T) {
f := setup(t)
c := mustDial(t, f)
c.recv(t) // initial snapshot
go func() {
st, err := c.conn.AcceptStream(context.Background())
if err != nil {
return
}
var open pb.ServerMessage
if transport.ReadMsg(st, &open, transport.DefaultMaxFrame) != nil {
return
}
_ = transport.WriteMsg(st, &pb.AgentMessage{Msg: &pb.AgentMessage_TcpOpened{
TcpOpened: &pb.TCPOpened{Ok: false, Error: "connection refused"}}})
}()
_, err := f.svc.OpenTCP(context.Background(), f.host.ID, "vm-1", 22)
require.Error(t, err)
assert.Contains(t, err.Error(), "connection refused")
}
func TestHelloPersistsHostFacts(t *testing.T) {
f := setup(t)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
conn, err := quic.DialAddr(ctx, f.addr, transport.ClientTLS(f.fp),
&quic.Config{MaxIdleTimeout: 5 * time.Second})
require.NoError(t, err)
t.Cleanup(func() { conn.CloseWithError(0, "") })
up, err := conn.OpenStreamSync(ctx)
require.NoError(t, err)
require.NoError(t, transport.WriteMsg(up, &pb.AgentMessage{Msg: &pb.AgentMessage_Hello{Hello: &pb.Hello{
HostId: f.host.ID, Credential: f.cred,
Facts: &pb.HostFacts{
OsId: "debian", OsPretty: "Debian GNU/Linux 12 (bookworm)", OsVersion: "12",
Kernel: "6.1.0-18-amd64", CpuModel: "AMD EPYC 7302P", Virt: "kvm",
},
}}}))
_, err = conn.AcceptStream(ctx) // down-stream opens only after the Hello is processed
require.NoError(t, err)
require.Eventually(t, func() bool {
h, err := f.st.GetHost(f.host.ID)
return err == nil && h.OSPretty == "Debian GNU/Linux 12 (bookworm)"
}, 2*time.Second, 20*time.Millisecond)
h, _ := f.st.GetHost(f.host.ID)
assert.Equal(t, "debian", h.OSID)
assert.Equal(t, "6.1.0-18-amd64", h.Kernel)
assert.Equal(t, "kvm", h.Virt)
}
func TestReportMetricsLandInRegistry(t *testing.T) {
f := setup(t)
c := mustDial(t, f)
c.recv(t) // initial snapshot
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{Report: &pb.Report{
Capacity: &pb.Capacity{Vcpus: 8},
Metrics: &pb.HostMetrics{UptimeS: 3600, Load1: 1.5, MemUsedMb: 2048, MemAvailableMb: 6144, DiskUsedGb: 20, DiskFreeGb: 80},
LastSeenEpoch: 2,
}}})
require.Eventually(t, func() bool {
st, ok := f.reg.Get(f.host.ID)
return ok && st.Metrics.UptimeS == 3600
}, 2*time.Second, 20*time.Millisecond)
st, _ := f.reg.Get(f.host.ID)
assert.InDelta(t, 1.5, st.Metrics.Load1, 0.001)
assert.Equal(t, int64(2048), st.Metrics.MemUsedMB)
assert.Equal(t, int64(80), st.Metrics.DiskFreeGB)
}
func TestUpgradeOffers(t *testing.T) {
s := &Service{offers: map[string]offer{}, now: time.Now}
s.OfferAgentUpgrade("h1", "v0.0.2", "https://eitri.sh/dl/v0.0.2/a.tar.gz", "ab")
up := s.offerFor("h1")
if up == nil || up.Version != "v0.0.2" {
t.Fatalf("offerFor = %v", up)
}
if s.offerFor("h2") != nil {
t.Fatal("offer must be per-host")
}
// A Hello reporting a DIFFERENT version keeps the offer.
s.clearOfferIfDone("h1", "v0.0.1")
if s.offerFor("h1") == nil {
t.Fatal("offer cleared too early")
}
// A Hello reporting the target clears it.
s.clearOfferIfDone("h1", "v0.0.2")
if s.offerFor("h1") != nil {
t.Fatal("offer not cleared at target version")
}
// Decommission clears unconditionally — a stale offer must not outlive its host.
s.OfferAgentUpgrade("h3", "v0.0.2", "u", "s")
s.ClearAgentUpgrade("h3")
if s.offerFor("h3") != nil {
t.Fatal("ClearAgentUpgrade must drop the offer")
}
}
// TestPendingAgentUpgradeAges pins the fact the console reads: an offer knows
// how long it has been standing, so "in flight" and "not landing" are
// distinguishable from outside the host.
func TestPendingAgentUpgradeAges(t *testing.T) {
clock := time.Date(2026, 8, 11, 9, 0, 0, 0, time.UTC)
s := &Service{offers: map[string]offer{}, now: func() time.Time { return clock }}
if _, _, ok := s.PendingAgentUpgrade("h1"); ok {
t.Fatal("a host with no offer has nothing pending")
}
s.OfferAgentUpgrade("h1", "v0.0.6", "https://eitri.sh/dl/v0.0.6/a.tar.gz", "ab")
v, age, ok := s.PendingAgentUpgrade("h1")
if !ok || v != "v0.0.6" || age != 0 {
t.Fatalf("fresh offer = %q %v %v", v, age, ok)
}
// The wait is measured at read time, not at offer time.
clock = clock.Add(4 * time.Minute)
if _, age, _ = s.PendingAgentUpgrade("h1"); age != 4*time.Minute {
t.Fatalf("age after 4m = %v", age)
}
// A second click is a fresh ask, and the wait it reports is the wait since
// that ask — otherwise the operator reads the age of an offer they replaced.
s.OfferAgentUpgrade("h1", "v0.0.6", "https://eitri.sh/dl/v0.0.6/a.tar.gz", "ab")
if _, age, _ = s.PendingAgentUpgrade("h1"); age != 0 {
t.Fatalf("re-offer did not restart the clock: age = %v", age)
}
// Convergence ends it: the agent reports the version it was offered.
s.clearOfferIfDone("h1", "v0.0.6")
if _, _, ok = s.PendingAgentUpgrade("h1"); ok {
t.Fatal("offer still pending after the agent reported the target version")
}
}
// TestHelloRefreshesProvisioner pins that a host which changes backend stops
// lying about itself on its next reconnect. The provisioner is recorded at
// enrollment and was only ever LOGGED on Hello, so a Mac enrolled before its
// backend existed kept reporting "inert" — a backend since deleted from the
// tree — to the console, the CLI and anyone reading the fleet.
func TestHelloRefreshesProvisioner(t *testing.T) {
f := setup(t)
before, err := f.st.GetHost(f.host.ID)
require.NoError(t, err)
require.Equal(t, "cloudhv", before.Provisioner, "fixture enrolls as cloudhv")
c, err := dialAs(t, f.addr, f.fp, f.host.ID, f.cred, "vfkit")
require.NoError(t, err)
c.recv(t) // initial snapshot: the Hello has been processed
require.Eventually(t, func() bool {
h, err := f.st.GetHost(f.host.ID)
return err == nil && h.Provisioner == "vfkit"
}, 2*time.Second, 20*time.Millisecond, "the host row must follow what the agent reports")
}
// TestReportRecordsTheHostsGuestSubnet pins the inversion end to end: the fleet
// records the subnet the HOST says its guests are on, rather than the host
// echoing back an allocation the fleet made for it. A Mac's guests live on
// vmnet's subnet, which no fleet allocation will ever contain.
func TestReportRecordsTheHostsGuestSubnet(t *testing.T) {
f := setup(t)
before, err := f.st.GetHost(f.host.ID)
require.NoError(t, err)
require.NotEqual(t, "192.168.64.0/24", before.BridgeCIDR)
c := mustDial(t, f)
c.recv(t)
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{GuestCidr: "192.168.64.0/24"}}})
require.Eventually(t, func() bool {
h, err := f.st.GetHost(f.host.ID)
return err == nil && h.BridgeCIDR == "192.168.64.0/24"
}, 2*time.Second, 20*time.Millisecond, "the fleet must record what the host reports")
// Empty is "no answer", never "no network": a host that cannot see its own
// network yet must not erase what it told us when it could.
c.send(t, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{
Report: &pb.Report{GuestCidr: ""}}})
require.Never(t, func() bool {
h, err := f.st.GetHost(f.host.ID)
return err == nil && h.BridgeCIDR != "192.168.64.0/24"
}, 500*time.Millisecond, 25*time.Millisecond, "an empty report must leave the record alone")
}
// TestSecondSessionEvictsTheFirst pins one host to one session. Before it, the
// hub silently displaced the elder's poke channel and the elder's handler
// returned without closing the connection: the agent on the other end kept
// writing reports into a stream with no reader, keepalives held the socket open
// indefinitely, and the host's last-seen froze until someone broke the
// connection by hand. The elder must SEE the close, and see it as transient so
// a live agent comes straight back rather than sitting out the auth backoff.
func TestSecondSessionEvictsTheFirst(t *testing.T) {
f := setup(t)
elder := mustDial(t, f)
elder.recv(t) // elder is the registered session
newer := mustDial(t, f)
newer.recv(t) // newer has taken over
require.NoError(t, elder.down.SetReadDeadline(time.Now().Add(5*time.Second)))
var discard pb.ServerMessage
err := transport.ReadMsg(elder.down, &discard, transport.DefaultMaxFrame)
require.Error(t, err, "the displaced session must be closed, not abandoned")
var appErr *quic.ApplicationError
require.True(t, errors.As(err, &appErr), "expected *quic.ApplicationError, got %T: %v", err, err)
assert.Equal(t, quic.ApplicationErrorCode(transport.CodeSuperseded), appErr.ErrorCode,
"the displaced agent must be told it was superseded, not that its credential is bad")
// The elder's teardown runs asynchronously and must take nothing of the
// newer's with it. Its console registration survives the whole window —
// deregistration is guarded on the map still holding the elder's own conn.
require.Never(t, func() bool {
f.svc.consoleMu.Lock()
defer f.svc.consoleMu.Unlock()
return f.svc.conns[f.host.ID] == nil
}, time.Second, 50*time.Millisecond,
"the evicted session's teardown deregistered the surviving one")
// And so does its hub subscription: a poke still reaches it.
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
f.hub.Poke(f.host.ID)
require.NoError(t, newer.down.SetReadDeadline(time.Now().Add(5*time.Second)))
snap := newer.recv(t).GetSnapshot()
require.NotNil(t, snap, "the surviving session must still be poked")
require.Len(t, snap.Vms, 1)
}
// TestSessionIdentityComesFromCredentialNotHello pins the claim that makes the
// Hello's host_id inert: a session is the host its CREDENTIAL names, never the
// host the Hello claims to be. The two are separable — the field is agent-sent
// and unauthenticated — so a valid credential paired with a foreign host_id is
// directly constructible, and is what an agent would send if it were lying.
//
// The stakes are a tenant's whole desired state. Under a server that trusts the
// field, the first snapshot pushed down this connection is the OTHER tenant's
// fleet, cloud-init included, and every fact this session reports lands on a
// host it does not own.
func TestSessionIdentityComesFromCredentialNotHello(t *testing.T) {
f := setup(t)
// A second host, under a second tenant, with a VM whose desired state must
// never cross to f.host's agent.
_, err := f.st.CreateTenantForIdentity("https://test-issuer", "beta-subject", "beta@test.local")
require.NoError(t, err)
betaTok, err := f.st.CreateEnrollmentToken("beta")
require.NoError(t, err)
hostB, err := f.st.RedeemEnrollmentToken(betaTok, store.EnrollFacts{
Name: "beta-host", OS: "linux", Arch: "amd64", Provisioner: "cloudhv", Remote: ""})
require.NoError(t, err)
require.NoError(t, f.st.CreateVM(store.VM{ID: "beta-vm", HostID: hostB.ID, Name: "secret",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
// f.cred is host A's. The Hello claims to be host B.
c, err := dialHello(t, f.addr, f.fp, &pb.Hello{
HostId: hostB.ID,
Credential: f.cred,
Facts: &pb.HostFacts{OsId: "spoofed"},
})
require.NoError(t, err, "the credential is valid, so the session must open — the claim is which identity it gets")
t.Cleanup(func() { c.conn.CloseWithError(0, "") })
snap := c.recv(t).GetSnapshot()
require.NotNil(t, snap)
for _, vm := range snap.Vms {
assert.NotEqual(t, "beta-vm", vm.VmId,
"the session snapshot must be the CREDENTIALED host's: a spoofed Hello.HostId leaked tenant beta's desired state to host %s's agent", f.host.ID)
}
require.Eventually(t, func() bool {
h, err := f.st.GetHost(f.host.ID)
return err == nil && h.OSID == "spoofed"
}, 2*time.Second, 10*time.Millisecond,
"Hello facts must land on the host the credential names")
hb, err := f.st.GetHost(hostB.ID)
require.NoError(t, err)
assert.Empty(t, hb.OSID, "the spoofed host must be untouched — it never connected")
_, onlineB := f.reg.Get(hostB.ID)
assert.False(t, onlineB, "a spoofed host_id must not make another host read as online")
}
// TestApplyReportRefusesAnotherHostsVM pins the rule the network_ip and host-key
// writes already hold, for the two facts that were left open: a host may only
// ever speak for the VMs it holds. Every id in a Report is agent-supplied, and
// a VM's id is not a secret the way an unguessable handle would be — cloud-init
// hands it to the guest as instance-id — so a VM id reaching a host in another
// tenant is an ordinary thing, not a break-in.
//
// The stakes are the two writes that trust it: the status line the owning
// tenant reads, and the hard delete that ends the teardown grace window early
// enough that a restore has nothing left to restore.
func TestApplyReportRefusesAnotherHostsVM(t *testing.T) {
f := setup(t)
// A second host, under a second tenant. It never holds vm1.
_, err := f.st.CreateTenantForIdentity("https://test-issuer", "beta-subject", "beta@test.local")
require.NoError(t, err)
betaTok, err := f.st.CreateEnrollmentToken("beta")
require.NoError(t, err)
hostB, err := f.st.RedeemEnrollmentToken(betaTok, store.EnrollFacts{
Name: "beta-host", OS: "linux", Arch: "amd64", Provisioner: "cloudhv", Remote: ""})
require.NoError(t, err)
require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
// The owning host reports the truth first, so the test measures a change
// against a known state rather than against an empty row.
f.svc.applyReport(f.host.ID, &pb.Report{Vms: []*pb.VMStatus{{
VmId: "vm1", PowerState: "running", Phase: "ready", Ip: "10.77.1.2"}}})
vm, err := f.st.GetVM("vm1")
require.NoError(t, err)
require.Equal(t, "ready", vm.Status)
require.Equal(t, "10.77.1.2", vm.AssignedIP)
// Host B names host A's VM.
f.svc.applyReport(hostB.ID, &pb.Report{Vms: []*pb.VMStatus{{
VmId: "vm1", PowerState: "running", Phase: "failed",
LastError: "seized", Ip: "10.9.9.9"}}})
vm, err = f.st.GetVM("vm1")
require.NoError(t, err)
assert.Equal(t, "ready", vm.Status, "a host that does not hold this VM must not write its status")
assert.Empty(t, vm.LastError, "nor the error line its tenant reads")
assert.Equal(t, "10.77.1.2", vm.AssignedIP, "nor the address every exposure and the gate aim at")
// The same rule on the reap: a tombstoned row is the teardown grace window,
// and only the host holding it may end that window.
require.NoError(t, f.st.TombstoneVM("vm1"))
f.svc.applyReport(hostB.ID, &pb.Report{Destroyed: []string{"vm1"}})
_, err = f.st.GetVM("vm1")
require.NoError(t, err, "a foreign ack must not reap the row out from under its restore window")
// The owning host's ack still reaps, so the guard costs the real path nothing.
f.svc.applyReport(f.host.ID, &pb.Report{Destroyed: []string{"vm1"}})
_, err = f.st.GetVM("vm1")
assert.ErrorIs(t, err, sql.ErrNoRows, "the host that holds the VM still reaps it")
}