1package broker
2
3import (
4 "testing"
5 "time"
6
7 "github.com/stretchr/testify/require"
8
9 "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/internal/ports"
11 "github.com/bnema/vev/internal/protocol/catalogue"
12)
13
14// TestRegistryIdleRefreshPublishesOnceAndSkipsUnchangedPersistence pins the
15// idle cost of a watched broker: a routine refresh of a reachable host
16// publishes once (its result, no transient Checking publication), and a result
17// whose durable content is unchanged does not rewrite the store.
18func TestRegistryIdleRefreshPublishesOnceAndSkipsUnchangedPersistence(t *testing.T) {
19 start := time.Unix(100, 0)
20 clock := newManualClock(start)
21 probe := newTestProbe(1)
22 store := newTestStore()
23 registry := newTestRegistry(t, 1, store, probe, clock)
24 registry.demandFreshForRemote = time.Second
25 reg := registration(t, "example.test", 1)
26 require.NoError(t, registry.setHosts(hostRecords(reg)))
27 startRegistry(t, registry)
28
29 answer := func(sessions []catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
30 t.Helper()
31 var call probeCall
32 require.Eventually(t, func() bool {
33 select {
34 case call = <-probe.calls:
35 return true
36 default:
37 clock.Advance(registry.demandFreshForRemote)
38 return false
39 }
40 }, registryTestWait, time.Millisecond)
41 call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
42 Availability: domain.RemoteAvailabilityReachable,
43 InventoryKnown: true,
44 Sessions: sessions,
45 }}
46 now := clock.Now()
47 return waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
48 return !host.Checking && host.LastSuccess.Equal(now)
49 })
50 }
51
52 first := []catalogue.RemoteCatalogSession{hostSession("work")}
53 answer(first)
54 stored := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
55 host, ok := snapshot.Find(reg.Endpoint)
56 return ok && len(host.Sessions) == 1
57 })
58 writes := store.writeCount()
59 revision := registry.Snapshot().Revision
60
61 for range 2 {
62 answer(first)
63 require.Equal(t, revision+1, registry.Snapshot().Revision, "one publication per routine refresh")
64 revision++
65 }
66
67 // A durable change is still written, and it is the only write since the
68 // first one: the idle refreshes never reached the store.
69 changed := []catalogue.RemoteCatalogSession{hostSession("work"), hostSession("play")}
70 answer(changed)
71 stored = waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
72 host, ok := snapshot.Find(reg.Endpoint)
73 return ok && len(host.Sessions) == 2
74 })
75 require.Equal(t, writes+1, store.writeCount())
76 require.Equal(t, registry.Snapshot().Revision, stored.Revision)
77}
78
79// TestRegistryExplicitProbeStillPublishesChecking keeps the visible refresh
80// state for requested observations and for hosts not confirmed reachable.
81func TestRegistryExplicitProbeStillPublishesChecking(t *testing.T) {
82 clock := newManualClock(time.Unix(100, 0))
83 probe := newTestProbe(1)
84 registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
85 reg := registration(t, "example.test", 1)
86 require.NoError(t, registry.setHosts(hostRecords(reg)))
87 startRegistry(t, registry)
88
89 // Unknown availability: the first scheduled observation is visible.
90 call := receiveCall(t, probe)
91 waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { return host.Checking })
92 call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
93 waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
94 return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
95 })
96
97 // Reachable host, explicit request: still visible.
98 registry.RequestProbe(reg.Endpoint)
99 call = receiveCall(t, probe)
100 waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { return host.Checking })
101 call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
102 waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { return !host.Checking })
103}