amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package dbreads
2
3import (
4 "context"
5 "errors"
6 "net"
7 "strings"
8 "sync/atomic"
9 "testing"
10 "time"
11
12 "github.com/benbjohnson/litestream"
13
14 "github.com/carlosframework/platform/internal/release"
15)
16
17// countingStore counts reads of request objects and lists.
18type countingStore struct {
19 release.Store
20 requestGets, lists atomic.Int64
21}
22
23func (s *countingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
24 if strings.HasSuffix(key, "/request.json") {
25 s.requestGets.Add(1)
26 }
27 return s.Store.Get(ctx, key)
28}
29
30func (s *countingStore) List(ctx context.Context, prefix string) ([]string, error) {
31 s.lists.Add(1)
32 return s.Store.List(ctx, prefix)
33}
34
35// Review Focus 2: a request issued at 10:59:50 is listed at 11:00:20 under
36// the previous hour's prefix, and picked.
37func TestPassFindsAnEntryFromThePreviousHour(t *testing.T) {
38 t.Parallel()
39 fx := newFixture(t, 1)
40 fx.now = time.Date(2026, 10, 22, 10, 59, 50, 0, time.UTC)
41 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
42 c.Now = func() time.Time { return time.Date(2026, 10, 22, 11, 0, 20, 0, time.UTC) }
43 })
44 req := fx.request("SELECT count(*) FROM t", "")
45 fx.submit(req)
46 a.Pass(context.Background())
47 a.Wait()
48 if _, _, r := fx.answer(req.ID); r.State != StateDone {
49 t.Fatalf("%+v", r)
50 }
51}
52
53func TestPassSkipsAnOutOfWindowEntryWithoutReadingIt(t *testing.T) {
54 t.Parallel()
55 fx := newFixture(t, 0)
56 cs := &countingStore{Store: fx.store}
57 fx.now = time.Now().UTC().Add(-3 * time.Minute).Truncate(time.Second)
58 fx.submit(fx.request("SELECT 1", ""))
59 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.Store = cs })
60 a.Pass(context.Background())
61 if cs.requestGets.Load() != 0 {
62 t.Fatal("an out-of-window request was read")
63 }
64}
65
66// T4, the box's half: a forged queue entry and an unsigned request are
67// refused, decided, and logged once.
68func TestPassRefusesForgedEntriesAndUnsignedRequests(t *testing.T) {
69 t.Parallel()
70 fx := newFixture(t, 1)
71 var logged atomic.Int64
72 a, w := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
73 c.Logf = func(format string, args ...any) {
74 if strings.Contains(format, "refused the request") {
75 logged.Add(1)
76 }
77 }
78 })
79 forged := fx.request("SELECT 1", "")
80 wire, _ := Sign(release.RootSigner(fx.rootPriv), forged)
81 fx.store.Put(context.Background(), RequestKey(forged.ID), wire, "")
82 issued, _ := forged.Issued()
83 fx.store.Put(context.Background(), QueueKey("uk", issued.Add(-time.Second), forged.ID), nil, "")
84 unsigned := fx.request("SELECT 1", "")
85 fx.store.Put(context.Background(), RequestKey(unsigned.ID), []byte(`{"payload":"e30=","sig":""}`), "")
86 fx.store.Put(context.Background(), QueueKey("uk", issued, unsigned.ID), nil, "")
87 a.Pass(context.Background())
88 a.Pass(context.Background())
89 a.Wait()
90 for _, id := range []string{forged.ID, unsigned.ID} {
91 if _, _, err := fx.store.Get(context.Background(), PickedKey(id)); err == nil {
92 t.Errorf("%s was picked", id)
93 }
94 }
95 if logged.Load() != 2 || w.dials.Load() != 0 {
96 t.Fatalf("logged %d refusals, dialed %d workers", logged.Load(), w.dials.Load())
97 }
98}
99
100// T4 and codex 5: the same envelope offered again, before and after the
101// agent restarts, is never run twice: picked holds.
102func TestReplayIsRefusedByPickedAcrossARestart(t *testing.T) {
103 t.Parallel()
104 fx := newFixture(t, 1)
105 a, _ := fx.agent("uk-1")
106 req := fx.request("SELECT count(*) FROM t", "")
107 fx.submit(req)
108 a.Pass(context.Background())
109 a.Wait()
110 a.Pass(context.Background()) // the same agent again
111 restarted, w := fx.agent("uk-1")
112 restarted.Pass(context.Background())
113 restarted.Wait()
114 if w.dials.Load() != 0 {
115 t.Fatal("a restarted agent ran a read it had already picked")
116 }
117}
118
119func TestTwoAgentsServeEachReadOnce(t *testing.T) {
120 t.Parallel()
121 fx := newFixture(t, 1)
122 a, wa := fx.agent("uk-1")
123 b, wb := fx.agent("uk-2")
124 var ids []string
125 for i := 0; i < 10; i++ {
126 req := fx.request("SELECT count(*) FROM t", "")
127 fx.submit(req)
128 ids = append(ids, req.ID)
129 }
130 for round := 0; round < 20; round++ {
131 done := make(chan struct{}, 2)
132 go func() { a.Pass(context.Background()); done <- struct{}{} }()
133 go func() { b.Pass(context.Background()); done <- struct{}{} }()
134 <-done
135 <-done
136 a.Wait()
137 b.Wait()
138 }
139 for _, id := range ids {
140 if _, _, r := fx.answer(id); r.State != StateDone {
141 t.Fatalf("%s: %+v", id, r)
142 }
143 }
144 if got := wa.dials.Load() + wb.dials.Load(); got != int64(len(ids)) {
145 t.Fatalf("%d worker connections for %d reads", got, len(ids))
146 }
147}
148
149// Plan review finding 14 (spec §7.1, D2): a draining box beside a healthy
150// one never wins a read, including one that starts draining during its
151// jitter. Mutation: drop the Draining check after the jitter; the draining
152// agent picks every read and this test is red. (The check before admission
153// only spares a reservation; the one after the jitter is the gate.)
154func TestADrainingBoxDoesNotPick(t *testing.T) {
155 t.Parallel()
156 fx := newFixture(t, 1)
157 var flip atomic.Bool
158 drain, dw := fx.agent("uk-drain", func(c *AgentConfig, _ *inProcWorkers) {
159 c.Draining = flip.Load
160 c.Jitter = func() time.Duration { flip.Store(true); return time.Millisecond }
161 })
162 healthy, _ := fx.agent("uk-1")
163 for i := 0; i < 5; i++ {
164 req := fx.request("SELECT 1", "")
165 fx.submit(req)
166 drain.Pass(context.Background()) // first: it passes the first check, then drains in its jitter
167 drain.Wait()
168 healthy.Pass(context.Background())
169 healthy.Wait()
170 if _, _, r := fx.answer(req.ID); r.Box != "uk-1" {
171 t.Fatalf("read %d served by %s", i, r.Box)
172 }
173 flip.Store(false)
174 }
175 if dw.dials.Load() != 0 {
176 t.Fatal("the draining box started a worker")
177 }
178}
179
180// Finding 11's box half: a request already cancelled is never picked.
181// Mutation: drop the cancel read in consider; the agent picks it.
182func TestPassSkipsACancelledRequest(t *testing.T) {
183 t.Parallel()
184 fx := newFixture(t, 1)
185 a, w := fx.agent("uk-1")
186 req := fx.request("SELECT 1", "")
187 fx.submit(req)
188 fx.store.Put(context.Background(), CancelKey(req.ID), nil, "")
189 a.Pass(context.Background())
190 a.Wait()
191 if _, _, err := fx.store.Get(context.Background(), PickedKey(req.ID)); err == nil || w.dials.Load() != 0 {
192 t.Fatal("a cancelled request was picked")
193 }
194}
195
196func TestPassNeedsTheCapability(t *testing.T) {
197 t.Parallel()
198 fx := newFixture(t, 0)
199 cs := &countingStore{Store: fx.store}
200 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.Store = cs })
201 a.probeOK.Store(false)
202 a.Pass(context.Background())
203 if cs.lists.Load() != 0 {
204 t.Fatal("an incapable box listed the queue")
205 }
206}
207
208func TestProbe(t *testing.T) {
209 t.Parallel()
210 for name, c := range map[string]struct {
211 self func() HelloReply
212 dial error
213 want bool
214 }{
215 "a sandboxed worker": {func() HelloReply { return HelloReply{V: ProtocolVersion, Interfaces: []string{"lo"}} }, nil, true},
216 "another version": {func() HelloReply { return HelloReply{V: 2, Interfaces: []string{"lo"}} }, nil, false},
217 "a network": {func() HelloReply { return HelloReply{V: 1, Interfaces: []string{"lo", "eth0"}} }, nil, false},
218 "a visible /data": {func() HelloReply { return HelloReply{V: 1, Interfaces: []string{"lo"}, DataEntries: 3} }, nil, false},
219 "a readable host.env": {func() HelloReply { return HelloReply{V: 1, Interfaces: []string{"lo"}, EtcReadable: true} }, nil, false},
220 "an observation that failed": {func() HelloReply {
221 return HelloReply{V: 1, Errors: []string{"interfaces: permission denied"}}
222 }, nil, false},
223 "no service": {nil, errors.New("connection refused"), false},
224 } {
225 t.Run(name, func(t *testing.T) {
226 t.Parallel()
227 fx := newFixture(t, 0)
228 a, _ := fx.agent("uk-1", func(cfg *AgentConfig, w *inProcWorkers) { w.opt.Self = c.self })
229 if c.dial != nil {
230 a.cfg.DialWorker = func(context.Context) (net.Conn, error) { return nil, c.dial }
231 }
232 a.probeOK.Store(!c.want)
233 a.Probe(context.Background())
234 if a.Capable() != c.want {
235 t.Fatalf("Capable = %v", a.Capable())
236 }
237 })
238 }
239}
240
241// Review Focus 5 and spec §7.1: a probe beside an orphan, or with no unit
242// listing, is skipped and the last result stands.
243func TestProbeIsSkippedWhileAdmissionIsBlocked(t *testing.T) {
244 t.Parallel()
245 for _, last := range []bool{true, false} {
246 fx := newFixture(t, 0)
247 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
248 c.Units = func(context.Context) (int, error) { return 1, nil } // an orphan
249 })
250 a.probeOK.Store(last)
251 a.Probe(context.Background())
252 if a.Capable() != last {
253 t.Fatalf("a skipped probe changed the capability from %v", last)
254 }
255 }
256}
257
258// T13, the unit half: a box running two reads keeps the capability; its
259// probe gets the third connection.
260func TestProbeOnTheThirdConnection(t *testing.T) {
261 t.Parallel()
262 fx := newFixture(t, 1)
263 hang := make(chan struct{})
264 a, w := fx.agent("uk-1", func(c *AgentConfig, w *inProcWorkers) {
265 w.opt.Restore = func(ctx context.Context, _ litestream.ReplicaClient, _ string, _ uint64) error {
266 <-hang
267 return errors.New("released")
268 }
269 })
270 for i := 0; i < 2; i++ {
271 fx.submit(fx.request("SELECT 1", ""))
272 }
273 a.Pass(context.Background())
274 deadline := time.Now().Add(5 * time.Second)
275 for w.live.Load() < 2 && time.Now().Before(deadline) {
276 time.Sleep(10 * time.Millisecond)
277 }
278 a.Probe(context.Background())
279 if !a.Capable() {
280 t.Fatal("a busy box lost the capability")
281 }
282 close(hang)
283 a.Wait()
284}
285
286func TestRunPassesOnANudge(t *testing.T) {
287 t.Parallel()
288 fx := newFixture(t, 1)
289 // The start probe's worker unit lingers as systemd's deactivation (and a
290 // loaded CI box) makes it: main's CI caught the nudge's pass meeting it
291 // as an orphan, refusing admission, and nothing passing again within
292 // the test's bound (ListEvery is 30 s).
293 a, w := fx.agent("uk-1", func(_ *AgentConfig, w *inProcWorkers) { w.linger = 300 * time.Millisecond })
294 ctx, cancel := context.WithCancel(context.Background())
295 defer cancel()
296 nudge := make(chan struct{}, 1)
297 go a.Run(ctx, nudge)
298 // A pass that meets that unit refuses admission by design (spec §5.7)
299 // and the read waits for a later pass. This test is about the nudge, so
300 // it nudges once the start probe has run and its unit is gone.
301 ready := time.Now().Add(10 * time.Second)
302 for !(w.dials.Load() >= 1 && w.live.Load() == 0 && a.Capable()) {
303 if time.Now().After(ready) {
304 t.Fatal("the start probe never finished")
305 }
306 time.Sleep(10 * time.Millisecond)
307 }
308 req := fx.request("SELECT count(*) FROM t", "")
309 fx.submit(req)
310 nudge <- struct{}{}
311 deadline := time.Now().Add(10 * time.Second)
312 for time.Now().Before(deadline) {
313 if _, _, err := fx.store.Get(context.Background(), ResultKey(req.ID)); err == nil {
314 return
315 }
316 time.Sleep(20 * time.Millisecond)
317 }
318 t.Fatal("a nudge did not serve the read")
319}
320
321// Plan review finding 29, through the real loop: a read whose worker never
322// starts asks for a probe while that worker's unit is still stopping, so
323// admission skips it; the probe stays owed and runs once the unit is gone,
324// and the capability is withdrawn well inside ProbeEvery. Mutation: Run
325// calls Probe once on probeAfter and drops it when skipped (no retry):
326// Capable stays true until ProbeEvery and this test is red.
327func TestAProbeSkippedWhileTheUnitLingersRunsOnceItIsGone(t *testing.T) {
328 t.Parallel()
329 fx := newFixture(t, 1)
330 a, w := fx.agent("uk-1", func(_ *AgentConfig, w *inProcWorkers) { w.linger = 700 * time.Millisecond })
331 ctx, cancel := context.WithCancel(context.Background())
332 nudge := make(chan struct{}, 1)
333 done := make(chan struct{})
334 go func() { a.Run(ctx, nudge); close(done) }()
335 defer func() { cancel(); <-done }()
336 until := func(what string, ok func() bool) {
337 t.Helper()
338 deadline := time.Now().Add(10 * time.Second)
339 for !ok() {
340 if time.Now().After(deadline) {
341 t.Fatalf("never: %s", what)
342 }
343 time.Sleep(20 * time.Millisecond)
344 }
345 }
346 // The start probe has run and its unit is gone.
347 until("the start probe", func() bool { return w.dials.Load() >= 1 && w.live.Load() == 0 && a.Capable() })
348 w.failStart.Store(true)
349 req := fx.request("SELECT 1", "")
350 fx.submit(req)
351 nudge <- struct{}{}
352 until("the read's result", func() bool {
353 _, _, err := fx.store.Get(context.Background(), ResultKey(req.ID))
354 return err == nil
355 })
356 until("the owed probe withdrawing the capability", func() bool { return !a.Capable() })
357}
358