1package daemon
2
3import (
4 "errors"
5 "io"
6 "log/slog"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/vev/internal/ports"
12 "github.com/stretchr/testify/require"
13)
14
15type catalogCacheStub struct {
16 load func() ([]ports.RemoteCatalogCacheEntry, error)
17 store func([]ports.RemoteCatalogCacheEntry) error
18}
19
20func (s catalogCacheStub) Load() ([]ports.RemoteCatalogCacheEntry, error) {
21 if s.load == nil {
22 return nil, nil
23 }
24 return s.load()
25}
26
27func (s catalogCacheStub) Store(entries []ports.RemoteCatalogCacheEntry) error {
28 if s.store == nil {
29 return nil
30 }
31 return s.store(entries)
32}
33
34func TestRemoteCatalogCacheStartup(t *testing.T) {
35 now := time.Unix(0, 42)
36 tests := []struct {
37 name string
38 load func() ([]ports.RemoteCatalogCacheEntry, error)
39 wantCache map[string]ports.RemoteCatalogCacheEntry
40 wantStatus map[string]remoteHostStatus
41 }{
42 {
43 name: "loads cached entries",
44 load: func() ([]ports.RemoteCatalogCacheEntry, error) {
45 return []ports.RemoteCatalogCacheEntry{{Host: "arch", FetchedAt: now, Sessions: []ports.RemoteCatalogSession{{Name: "work", State: "up"}}}}, nil
46 },
47 wantCache: map[string]ports.RemoteCatalogCacheEntry{"arch": {Host: "arch", FetchedAt: now, Sessions: []ports.RemoteCatalogSession{{Name: "work", State: "up"}}}},
48 wantStatus: map[string]remoteHostStatus{"arch": remoteHostCached},
49 },
50 {
51 name: "invalid cache starts empty",
52 load: func() ([]ports.RemoteCatalogCacheEntry, error) {
53 return nil, errors.New("invalid cache")
54 },
55 wantCache: map[string]ports.RemoteCatalogCacheEntry{},
56 wantStatus: map[string]remoteHostStatus{},
57 },
58 }
59
60 for _, tt := range tests {
61 t.Run(tt.name, func(t *testing.T) {
62 d := New(nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil)), WithRemoteDiscovery(nil, nil, catalogCacheStub{load: tt.load}, nil, ports.RemoteTransportUDP))
63
64 d.remoteCatalog.mu.Lock()
65 defer d.remoteCatalog.mu.Unlock()
66 require.Equal(t, tt.wantCache, d.remoteCatalog.cache)
67 require.Equal(t, tt.wantStatus, d.remoteCatalog.status)
68 })
69 }
70}
71
72func TestRemoteCatalogCacheStoreSerializesLockFreeIO(t *testing.T) {
73 var d *Daemon
74 entered := make(chan struct{}, 2)
75 release := make(chan struct{})
76 var active, maxActive int
77 var activeMu sync.Mutex
78
79 cache := catalogCacheStub{store: func(entries []ports.RemoteCatalogCacheEntry) error {
80 if !d.mu.TryLock() {
81 return errors.New("daemon lock held during cache I/O")
82 }
83 d.mu.Unlock()
84 if !d.remoteCatalog.mu.TryLock() {
85 return errors.New("remote catalog lock held during cache I/O")
86 }
87 d.remoteCatalog.mu.Unlock()
88
89 activeMu.Lock()
90 active++
91 if active > maxActive {
92 maxActive = active
93 }
94 activeMu.Unlock()
95 entered <- struct{}{}
96 <-release
97 activeMu.Lock()
98 active--
99 activeMu.Unlock()
100 return nil
101 }}
102 d = New(nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil)), WithRemoteDiscovery(nil, nil, cache, nil, ports.RemoteTransportUDP))
103 d.remoteCatalog.mu.Lock()
104 d.remoteCatalog.cache["arch"] = ports.RemoteCatalogCacheEntry{Host: "arch", FetchedAt: time.Unix(0, 1), Sessions: []ports.RemoteCatalogSession{{Name: "one", State: "up"}}}
105 d.remoteCatalog.mu.Unlock()
106
107 done := make(chan struct{}, 2)
108 go func() { d.persistRemoteCatalogCache(); done <- struct{}{} }()
109 <-entered
110 d.remoteCatalog.mu.Lock()
111 d.remoteCatalog.cache["arch"] = ports.RemoteCatalogCacheEntry{Host: "arch", FetchedAt: time.Unix(0, 2), Sessions: []ports.RemoteCatalogSession{{Name: "two", State: "up"}}}
112 d.remoteCatalog.mu.Unlock()
113 go func() { d.persistRemoteCatalogCache(); done <- struct{}{} }()
114
115 select {
116 case <-entered:
117 t.Fatal("concurrent cache stores were not serialized")
118 case <-time.After(10 * time.Millisecond):
119 }
120 close(release)
121 <-entered
122 <-done
123 <-done
124
125 activeMu.Lock()
126 defer activeMu.Unlock()
127 require.Equal(t, 1, maxActive)
128}