1package swarm
2
3import (
4 "errors"
5 "fmt"
6 "os"
7 "path/filepath"
8 "runtime"
9 "sync"
10 "syscall"
11 "testing"
12 "time"
13)
14
15// shortSocketDir returns a tempdir short enough that we can append
16// "in.sock" and still fit under the unix-socket path cap. The Go
17// runner's t.TempDir() lives under a long /var/folders/... path on
18// macOS which blows the 104-byte limit by itself, so the inbox tests
19// can't use it.
20func shortSocketDir(t *testing.T) string {
21 t.Helper()
22 if runtime.GOOS == "windows" {
23 t.Skip("swarm inbox transport uses Unix-domain sockets; Windows support will use a named-pipe backend")
24 }
25 dir, err := os.MkdirTemp("/tmp", "zot-in-")
26 if err != nil {
27 t.Fatalf("mkdir: %v", err)
28 }
29 t.Cleanup(func() { os.RemoveAll(dir) })
30 return dir
31}
32
33// TestInboxRoundTrip is the happy path: child listens, supervisor
34// dials and sends three lines, child receives them in order, and
35// closing both sides is clean. Locks in the line-framed protocol.
36func TestInboxRoundTrip(t *testing.T) {
37 path := filepath.Join(shortSocketDir(t), "in.sock")
38 ln, err := Listen(path)
39 if err != nil {
40 t.Fatalf("Listen: %v", err)
41 }
42 defer ln.Close()
43
44 got := collectFor(t, ln.Lines(), 3, time.Second)
45 inbox := NewInbox(path)
46 defer inbox.Close()
47
48 want := []string{"user hello", "user world", "cancel"}
49 for _, msg := range want {
50 if err := inbox.SendInput(msg); err != nil {
51 t.Fatalf("SendInput(%q): %v", msg, err)
52 }
53 }
54
55 gotLines := <-got
56 if len(gotLines) != len(want) {
57 t.Fatalf("got %d lines, want %d: %v", len(gotLines), len(want), gotLines)
58 }
59 for i, w := range want {
60 if gotLines[i] != w {
61 t.Errorf("line %d = %q; want %q", i, gotLines[i], w)
62 }
63 }
64}
65
66// TestInboxNotReady regression-tests the explicit "child hasn't
67// opened the socket yet" case. The TUI needs an error it can show
68// as "agent not listening yet" rather than a generic dial failure.
69func TestInboxNotReady(t *testing.T) {
70 path := filepath.Join(shortSocketDir(t), "in.sock")
71 inbox := NewInbox(path)
72 defer inbox.Close()
73 err := inbox.SendInput("user hi")
74 if err == nil {
75 t.Fatal("SendInput on missing socket: want error")
76 }
77 if !errors.Is(err, ErrNotReady) {
78 t.Fatalf("SendInput err = %v; want ErrNotReady", err)
79 }
80}
81
82// TestIsNoListenerErrMatchesECONNREFUSED documents the contract
83// that the user's screenshot regression depends on: any dial error
84// whose root cause is "the kernel refused the connect because nobody
85// is listening" must fold into ErrNotReady at the SendInput layer,
86// so the TUI shows "agent not accepting input (press R to resume)"
87// instead of leaking the unix-socket path. We pattern-match via
88// errors.Is on syscall.ECONNREFUSED rather than string-matching the
89// error text so platform variants all hit the same branch.
90func TestIsNoListenerErrMatchesECONNREFUSED(t *testing.T) {
91 if !isNoListenerErr(syscall.ECONNREFUSED) {
92 t.Error("isNoListenerErr(ECONNREFUSED) = false; want true")
93 }
94 // Wrapped via fmt.Errorf — common form of the dial error.
95 if !isNoListenerErr(fmt.Errorf("dial unix /tmp/x: %w", syscall.ECONNREFUSED)) {
96 t.Error("isNoListenerErr(wrapped) = false; want true")
97 }
98 if isNoListenerErr(syscall.ECONNRESET) {
99 t.Error("isNoListenerErr(ECONNRESET) = true; want false (unrelated errno)")
100 }
101 if isNoListenerErr(nil) {
102 t.Error("isNoListenerErr(nil) = true; want false")
103 }
104}
105
106// TestInboxNewSupervisorPreemptsOld documents that only one
107// supervisor connection at a time is honoured. The swarm design
108// assumes a single parent zot owns each agent; if a second
109// parent dials, the listener boots the first one. The first
110// supervisor's already-delivered messages still land in the
111// channel — what we're guarding against is silent dual ownership,
112// not message loss.
113func TestInboxNewSupervisorPreemptsOld(t *testing.T) {
114 path := filepath.Join(shortSocketDir(t), "in.sock")
115 ln, err := Listen(path)
116 if err != nil {
117 t.Fatalf("Listen: %v", err)
118 }
119 defer ln.Close()
120
121 first := NewInbox(path)
122 if err := first.SendInput("user one"); err != nil {
123 t.Fatalf("first send: %v", err)
124 }
125 // Wait for the listener to absorb "user one" before dialing the
126 // second supervisor; otherwise we race against the connection
127 // preemption and "user one" can be dropped (the first conn is
128 // closed by Accept before its read loop processes the byte).
129 waitN(t, ln.Lines(), 1, time.Second)
130
131 second := NewInbox(path)
132 defer second.Close()
133 if err := second.SendInput("user two"); err != nil {
134 t.Fatalf("second send: %v", err)
135 }
136
137 secondLines := waitN(t, ln.Lines(), 1, time.Second)
138 if len(secondLines) != 1 || secondLines[0] != "user two" {
139 t.Fatalf("second supervisor not received: %v", secondLines)
140 }
141
142 // The original supervisor's connection was preempted by the
143 // listener accepting the second dial. The next SendInput on the
144 // first handle sees a broken pipe and drops the cached connection;
145 // the call after that redials successfully. Both behaviours
146 // matter: the broken-pipe error must surface (so a real
147 // supervisor knows it lost ownership) AND the next call must
148 // recover so a flaky reconnect doesn't permanently wedge the
149 // inbox.
150 if err := first.SendInput("user three"); err == nil {
151 t.Fatal("expected broken-pipe error on stale connection")
152 }
153 if err := first.SendInput("user four"); err != nil {
154 t.Fatalf("first redial after eviction: %v", err)
155 }
156}
157
158func waitN(t *testing.T, ch <-chan string, n int, timeout time.Duration) []string {
159 t.Helper()
160 var got []string
161 deadline := time.After(timeout)
162 for len(got) < n {
163 select {
164 case msg := <-ch:
165 got = append(got, msg)
166 case <-deadline:
167 return got
168 }
169 }
170 return got
171}
172
173// collectFor consumes up to n items from ch within timeout and
174// pushes the slice onto the returned channel. Used by the tests
175// above so they don't race against the listener's accept goroutine.
176func collectFor(t *testing.T, ch <-chan string, n int, timeout time.Duration) <-chan []string {
177 t.Helper()
178 out := make(chan []string, 1)
179 go func() {
180 var got []string
181 deadline := time.After(timeout)
182 for len(got) < n {
183 select {
184 case msg, ok := <-ch:
185 if !ok {
186 out <- got
187 return
188 }
189 got = append(got, msg)
190 case <-deadline:
191 out <- got
192 return
193 }
194 }
195 out <- got
196 }()
197 return out
198}
199
200// TestInboxConcurrentSends serialises the SendInput call so two
201// goroutines writing at once still produce two whole lines, never
202// a tearing of bytes between them.
203func TestInboxConcurrentSends(t *testing.T) {
204 path := filepath.Join(shortSocketDir(t), "in.sock")
205 ln, err := Listen(path)
206 if err != nil {
207 t.Fatalf("Listen: %v", err)
208 }
209 defer ln.Close()
210 inbox := NewInbox(path)
211 defer inbox.Close()
212
213 var wg sync.WaitGroup
214 const n = 50
215 wg.Add(n)
216 for i := 0; i < n; i++ {
217 go func(i int) {
218 defer wg.Done()
219 if err := inbox.SendInput("user msg"); err != nil {
220 t.Errorf("send %d: %v", i, err)
221 }
222 }(i)
223 }
224 wg.Wait()
225
226 collected := collectFor(t, ln.Lines(), n, 2*time.Second)
227 got := <-collected
228 if len(got) != n {
229 t.Fatalf("got %d lines, want %d", len(got), n)
230 }
231 for i, g := range got {
232 if g != "user msg" {
233 t.Fatalf("line %d garbled: %q", i, g)
234 }
235 }
236}