internal/server/syncsvc/tracker_test.go
Ref: Size: 11.7 KiB History
package syncsvc
import (
"errors"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
)
// write is a tiny helper that runs writeThrough with a SUCCESSFUL durable write
// that STORES the address it was given, returning whether a write actually
// happened (the report could still change the row).
func (t *statusTracker) write(vmID, status, lastErr, ip string) bool {
wrote := false
_ = t.writeThrough(vmID, status, lastErr, ip, func() (string, error) { wrote = true; return ip, nil })
return wrote
}
func TestStatusTrackerDecide(t *testing.T) {
tr := newStatusTracker()
// First call for a VM must write.
if !tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("first call must write")
}
// Identical repeat must not write.
if tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("identical repeat must not write")
}
// Changed phase must write.
if !tr.write("vm1", "failed", "", "10.0.0.1") {
t.Fatal("changed phase must write")
}
// Changed lastError must write.
if !tr.write("vm1", "failed", "boom", "10.0.0.1") {
t.Fatal("changed lastError must write")
}
// Changed non-empty ip must write.
if !tr.write("vm1", "failed", "boom", "10.0.0.2") {
t.Fatal("changed non-empty ip must write")
}
// ip="" with same phase/err keeps the prior ip → unchanged → no write.
if tr.write("vm1", "failed", "boom", "") {
t.Fatal("empty ip keeps prior; unchanged must not write")
}
}
func TestStatusTrackerEmptyThenNonEmptyIP(t *testing.T) {
tr := newStatusTracker()
// First report has NO ip: effective ip is empty.
if !tr.write("vm1", "ready", "", "") {
t.Fatal("first call must write")
}
// Same phase/err, still empty ip: unchanged (empty keeps prior empty).
if tr.write("vm1", "ready", "", "") {
t.Fatal("repeat empty-ip must not write")
}
// Later a non-empty ip arrives with same phase/err: this is a change.
if !tr.write("vm1", "ready", "", "10.0.0.5") {
t.Fatal("empty→non-empty ip must write")
}
// Now an empty ip keeps that prior non-empty ip → unchanged.
if tr.write("vm1", "ready", "", "") {
t.Fatal("empty ip after non-empty keeps prior; must not write")
}
}
func TestStatusTrackerFailedWriteNotCached(t *testing.T) {
tr := newStatusTracker()
// The durable write FAILS: writeThrough must surface the error and NOT cache.
sentinel := errors.New("record rejected")
if err := tr.writeThrough("vm1", "ready", "", "10.0.0.1", func() (string, error) {
return "", sentinel
}); err != sentinel {
t.Fatalf("failed write must surface its error, got %v", err)
}
// Because the write failed (uncached), the next report must still attempt it —
// a failed write must never suppress the retry.
attempted := false
if err := tr.writeThrough("vm1", "ready", "", "10.0.0.1", func() (string, error) {
attempted = true
return "10.0.0.1", nil
}); err != nil {
t.Fatalf("retry write errored: %v", err)
}
if !attempted {
t.Fatal("failed (uncached) write must not suppress the retry")
}
// After that successful write, an identical report must not write again.
if tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("after a successful write an identical report must not write")
}
}
// TestStatusTrackerDroppedAddressNotCached pins the half of the invariant that a
// silent guard broke: the cache holds what the STORE says it stored, never what
// the agent merely reported. A store that drops an address says so by returning
// nothing, and the tracker must then leave the cache where the row is — because
// a cache that remembers a rejected address treats every later report of it as
// "unchanged" and suppresses the write forever, which is how a guest could run
// for days with a blank assigned_ip while its agent reported the address on
// every single tick.
func TestStatusTrackerDroppedAddressNotCached(t *testing.T) {
tr := newStatusTracker()
dropped := func() (string, error) { return "", nil } // the store rejected it
require.NoError(t, tr.writeThrough("vm1", "ready", "", "169.254.11.2", dropped))
tr.mu.Lock()
cached := tr.last["vm1"].ip
tr.mu.Unlock()
require.Empty(t, cached, "a rejected address must not be cached as stored")
// The same rejected address again still reaches the store: nothing has been
// written, so there is nothing to skip.
attempted := false
require.NoError(t, tr.writeThrough("vm1", "ready", "", "169.254.11.2", func() (string, error) {
attempted = true
return "", nil
}))
require.True(t, attempted, "a repeat of a rejected address must still attempt the write")
// And the address that finally works lands, rather than being mistaken for a
// repeat of something already stored.
if !tr.write("vm1", "ready", "", "10.0.0.5") {
t.Fatal("a good address after a rejected one must write")
}
if tr.write("vm1", "ready", "", "10.0.0.5") {
t.Fatal("once stored, the same address must not write again")
}
}
// parkedWriter is the shape both write-through atomicity tests need: a durable
// write that announces it has started and then holds still until released, so a
// test can apply a second write for the same key at the one moment the two
// could interleave. The lock either keeps the second writer out of its write —
// which is the invariant — or it does not, and the second writer says so
// immediately. It is never a question of which goroutine wins a race: under a
// broken lock the second writer is runnable and blocked on nothing.
type parkedWriter struct {
inside chan struct{} // closed once this writer is inside its durable write
release chan struct{} // closed by the test to let it finish
}
func newParkedWriter() *parkedWriter {
return &parkedWriter{inside: make(chan struct{}), release: make(chan struct{})}
}
// entered reports whether a writer got into its durable write within a window
// long enough that only a starved goroutine could miss it.
func entered(inside <-chan struct{}) bool {
select {
case <-inside:
return true
case <-time.After(250 * time.Millisecond):
return false
}
}
// TestStatusTrackerWriteThroughAtomic pins that decide→write→commit runs under
// one lock per VM. Concurrent reports for the same host — an agent reconnect,
// where the old and new sessions both deliver a report — must not interleave
// the durable write and the cache update, because a cache left disagreeing with
// the row treats every later identical report as "unchanged" and suppresses it
// forever.
//
// The exclusion is asserted directly rather than sampled: -race cannot see this
// (the tracker's own mutex covers every shared access, so an unlocked write is
// still race-clean), and an end-state check over many random pairs is both a
// coin flip per run and self-healing — a divergence is papered over by the next
// write of the other status, so only one among the final few survives to be
// seen.
func TestStatusTrackerWriteThroughAtomic(t *testing.T) {
tr := newStatusTracker()
first := newParkedWriter()
secondInside := make(chan struct{})
var mu sync.Mutex
lastWritten := "" // status of the most recent successful durable write
go func() {
_ = tr.writeThrough("vm1", "ready", "", "10.0.0.1", func() (string, error) {
close(first.inside)
<-first.release
mu.Lock()
lastWritten = "ready"
mu.Unlock()
return "10.0.0.1", nil
})
}()
<-first.inside
done := make(chan struct{})
go func() {
defer close(done)
_ = tr.writeThrough("vm1", "failed", "", "10.0.0.1", func() (string, error) {
close(secondInside)
mu.Lock()
lastWritten = "failed"
mu.Unlock()
return "10.0.0.1", nil
})
}()
if entered(secondInside) {
close(first.release)
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")
}
close(first.release)
<-done
tr.mu.Lock()
cached := tr.last["vm1"].status
tr.mu.Unlock()
mu.Lock()
dbLast := lastWritten
mu.Unlock()
require.Equal(t, dbLast, cached, "cache must equal the last durable write")
}
// TestNetTrackerWriteThroughAtomic is the same invariant for the per-host facts
// (subnet, host address). Two connections for one host — the reconnect again —
// must not interleave: a cache holding a value the row never took makes every
// later report of the true value read as "unchanged", and the fleet keeps
// serving the stale subnet for as long as the process lives.
func TestNetTrackerWriteThroughAtomic(t *testing.T) {
tr := newNetTracker()
first := newParkedWriter()
secondInside := make(chan struct{})
var mu sync.Mutex
lastWritten := ""
go func() {
_ = tr.writeThrough("h1", "10.77.1.0/24", func() error {
close(first.inside)
<-first.release
mu.Lock()
lastWritten = "10.77.1.0/24"
mu.Unlock()
return nil
})
}()
<-first.inside
done := make(chan struct{})
go func() {
defer close(done)
_ = tr.writeThrough("h1", "192.168.64.0/24", func() error {
close(secondInside)
mu.Lock()
lastWritten = "192.168.64.0/24"
mu.Unlock()
return nil
})
}()
if entered(secondInside) {
close(first.release)
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")
}
close(first.release)
<-done
tr.mu.Lock()
cached := tr.last["h1"]
tr.mu.Unlock()
mu.Lock()
dbLast := lastWritten
mu.Unlock()
require.Equal(t, dbLast, cached, "cache must equal the last durable write")
}
func TestStatusTrackerForget(t *testing.T) {
tr := newStatusTracker()
if !tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("first call must write")
}
tr.forget("vm1")
// After forget the VM is unknown again, so the same report writes afresh.
if !tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("after forget the same report must write again")
}
}
func TestStatusTrackerPerVMKeys(t *testing.T) {
tr := newStatusTracker()
if !tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("vm1 first write")
}
// A different VM with the same triple is independent → must write.
if !tr.write("vm2", "ready", "", "10.0.0.1") {
t.Fatal("vm2 is a distinct key and must write")
}
if tr.write("vm1", "ready", "", "10.0.0.1") {
t.Fatal("vm1 unchanged must not write")
}
}
// TestNetTrackerWritesOnceThenStaysQuiet pins the guard that makes the steady
// state free: every host reports its subnet every tick forever, and the store
// runs on one connection, so an unguarded write spends a round trip per host
// per tick for the life of the fleet.
func TestNetTrackerWritesOnceThenStaysQuiet(t *testing.T) {
tr := newNetTracker()
writes := 0
report := func(host, cidr string) error {
return tr.writeThrough(host, cidr, func() error { writes++; return nil })
}
require.NoError(t, report("h1", "10.77.1.0/24"))
require.Equal(t, 1, writes, "the first report writes")
require.NoError(t, report("h1", "10.77.1.0/24"))
require.Equal(t, 1, writes, "a repeat writes nothing")
require.NoError(t, report("h1", "192.168.64.0/24"))
require.Equal(t, 2, writes, "a changed subnet writes")
// Hosts are independent.
require.NoError(t, report("h2", "192.168.64.0/24"))
require.Equal(t, 3, writes, "a different host is a different key")
// A REJECTED write is not remembered, so the next report retries it rather
// than being suppressed by a value that never reached the row.
sentinel := errors.New("rejected")
require.ErrorIs(t, tr.writeThrough("h3", "junk", func() error { return sentinel }), sentinel)
attempted := false
require.NoError(t, tr.writeThrough("h3", "junk", func() error { attempted = true; return nil }))
require.True(t, attempted, "a failed write must not suppress the retry")
// Forgetting a host writes it afresh: the row may have changed while it was
// disconnected.
tr.forget("h1")
require.NoError(t, report("h1", "192.168.64.0/24"))
require.Equal(t, 4, writes)
}