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 "bytes"
5 "context"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "io"
11 "os"
12 "reflect"
13 "strings"
14 "sync"
15 "sync/atomic"
16 "testing"
17 "time"
18
19 "github.com/benbjohnson/litestream"
20 lsfile "github.com/benbjohnson/litestream/file"
21 "github.com/superfly/ltx"
22
23 "github.com/carlosframework/platform/internal/dbdoor"
24 "github.com/carlosframework/platform/internal/provision"
25 "github.com/carlosframework/platform/internal/release"
26)
27
28func serveOne(t *testing.T, a *Agent, fx *fixture, req Request) ([]byte, Summary, Result) {
29 t.Helper()
30 e := fx.submit(req)
31 a.consider(context.Background(), e)
32 a.Wait()
33 return fx.answer(req.ID)
34}
35
36func TestReadServesAQuery(t *testing.T) {
37 t.Parallel()
38 fx := newFixture(t, 3)
39 a, w := fx.agent("uk-1")
40 rows, s, r := serveOne(t, a, fx, fx.request("SELECT count(*) AS n FROM t", ""))
41 if r.State != StateDone || r.Code != "" || string(rows) != "n\n30\n" || r.Box != "uk-1" {
42 t.Fatalf("result %+v, rows %q", r, rows)
43 }
44 plan, err := dbdoor.PlanFrom(context.Background(), lsfile.NewReplicaClient(fx.replica), time.Time{})
45 if err != nil || s.TXID != plan.TXID || !s.At.Equal(plan.At) {
46 t.Fatalf("summary %+v, plan %+v %v", s, plan, err)
47 }
48 if w.dials.Load() != 1 {
49 t.Fatalf("%d worker connections for one read", w.dials.Load())
50 }
51}
52
53// vanishing makes full-file opens of the first planned file answer
54// os.ErrNotExist, `times` times: litestream's compaction removing an L0
55// file under the plan.
56type vanishing struct {
57 dbdoor.ReplicaReader
58 mu sync.Mutex
59 times int
60}
61
62func (v *vanishing) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, off, size int64) (io.ReadCloser, error) {
63 v.mu.Lock()
64 gone := off == 0 && size == 0 && v.times > 0
65 if gone {
66 v.times--
67 }
68 v.mu.Unlock()
69 if gone {
70 return nil, os.ErrNotExist
71 }
72 return v.ReplicaReader.OpenLTXFile(ctx, level, lo, hi, off, size)
73}
74
75// Review Focus 3: one quiet re-plan, then changed-twice, and no chunk.
76func TestReadRePlansOnceThenSaysChangedTwice(t *testing.T) {
77 t.Parallel()
78 for _, c := range []struct {
79 times int
80 want string
81 }{{1, ""}, {2, CodeChangedTwice}} {
82 fx := newFixture(t, 2)
83 v := &vanishing{ReplicaReader: lsfile.NewReplicaClient(fx.replica), times: c.times}
84 a, _ := fx.agent("uk-1", func(cfg *AgentConfig, _ *inProcWorkers) {
85 cfg.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return v, nil }
86 })
87 req := fx.request("SELECT count(*) FROM t", "")
88 _, _, r := serveOne(t, a, fx, req)
89 if r.Code != c.want {
90 t.Fatalf("vanished %d times: %+v", c.times, r)
91 }
92 if c.want != "" && len(fx.chunkKeys(req.ID)) != 0 {
93 t.Fatal("a chunk exists for a read that never ran")
94 }
95 }
96}
97
98// Plan review finding 10: after a vanished file, the retry waits for the
99// first worker's unit to stop before it dials again, and keeps that unit's
100// reservation until then. 12 GiB of 20 fits one floor at a time.
101// Mutations: drop the settle call in read (the second dial sees a live unit,
102// and the retired floor makes the new one no-room); or drop retry's move of
103// the old reservation (TestAdmissionRetryHoldsTheOldReservation goes red).
104func TestRetryWaitsForTheFirstWorkerToStop(t *testing.T) {
105 t.Parallel()
106 fx := newFixture(t, 2)
107 v := &vanishing{ReplicaReader: lsfile.NewReplicaClient(fx.replica), times: 1}
108 a, w := fx.agent("uk-1", func(c *AgentConfig, w *inProcWorkers) {
109 c.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return v, nil }
110 c.Free = func() (uint64, uint64, error) { return 12 << 30, 20 << 30, nil }
111 w.linger = 300 * time.Millisecond
112 })
113 _, _, r := serveOne(t, a, fx, fx.request("SELECT count(*) FROM t", ""))
114 if r.State != StateDone {
115 t.Fatalf("result %+v", r)
116 }
117 w.mu.Lock()
118 defer w.mu.Unlock()
119 if len(w.liveAtDial) != 2 || w.liveAtDial[1] != 0 {
120 t.Fatalf("live units at each dial: %v; the retry dialed beside the first worker", w.liveAtDial)
121 }
122}
123
124// Plan review finding 15: a worker that ends before ready (its service
125// failed after the socket accepted) asks for a probe at once. Mutation:
126// drop the askProbe after a pre-ready sendFile or readFrame failure;
127// probeAfter stays empty and this test is red.
128func TestAWorkerThatNeverStartsAsksForAProbe(t *testing.T) {
129 t.Parallel()
130 fx := newFixture(t, 1)
131 a, _ := fx.agent("uk-1", func(_ *AgentConfig, w *inProcWorkers) { w.failStart.Store(true) })
132 _, _, r := serveOne(t, a, fx, fx.request("SELECT 1", ""))
133 if r.Code != CodeInterrupted || len(a.probeAfter) != 1 {
134 t.Fatalf("result %+v, probe asked %d", r, len(a.probeAfter))
135 }
136}
137
138// stageHook changes the record at one stage of the read: planning (the
139// first listing), download (the first full-file open), or after the last
140// byte has been read from the bucket (EOF of the last planned file).
141type stageHook struct {
142 dbdoor.ReplicaReader
143 stage, lastSpan string
144 once sync.Once
145 change func()
146}
147
148func (h *stageHook) LTXFiles(ctx context.Context, level int, seek ltx.TXID, m bool) (ltx.FileIterator, error) {
149 if h.stage == "planning" {
150 h.once.Do(h.change)
151 }
152 return h.ReplicaReader.LTXFiles(ctx, level, seek, m)
153}
154
155func (h *stageHook) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, off, size int64) (io.ReadCloser, error) {
156 rc, err := h.ReplicaReader.OpenLTXFile(ctx, level, lo, hi, off, size)
157 if err != nil || off != 0 || size != 0 {
158 return rc, err
159 }
160 switch {
161 case h.stage == "download":
162 h.once.Do(h.change)
163 case h.stage == "after the last byte" && dbdoor.Span(uint64(lo), uint64(hi)) == h.lastSpan:
164 return &eofHook{ReadCloser: rc, at: func() { h.once.Do(h.change) }}, nil
165 }
166 return rc, nil
167}
168
169type eofHook struct {
170 io.ReadCloser
171 at func()
172}
173
174func (e *eofHook) Read(p []byte) (int, error) {
175 n, err := e.ReadCloser.Read(p)
176 if err == io.EOF {
177 e.at()
178 }
179 return n, err
180}
181
182// T9: a record change at every stage ends with its code, and no chunk is
183// ever written (codex 1).
184func TestRecordChangeAtEveryStageWritesNoChunk(t *testing.T) {
185 t.Parallel()
186 no := false
187 changes := map[string]struct {
188 change func(fx *fixture)
189 want string
190 }{
191 "moved": {func(fx *fixture) { r := fx.record(); r.Region = "ie"; fx.putRecord(r) }, CodeRecordChanged},
192 "moving": {func(fx *fixture) {
193 r := fx.record()
194 r.Move = &provision.MoveState{ID: "mv_x", To: "ie", State: "requested"}
195 fx.putRecord(r)
196 }, CodeMoving},
197 "recreated in the account": {func(fx *fixture) { r := fx.record(); r.CreatedAt = r.CreatedAt.Add(time.Hour); fx.putRecord(r) }, CodeRecordChanged},
198 "recreated in another account": {func(fx *fixture) {
199 created := fx.record().CreatedAt
200 fx.store.Delete(context.Background(), provision.RecordKey(fxAccount, fxApp, fxHost))
201 fx.putRecordUnder("bac", provision.Record{Region: "uk", CreatedAt: created})
202 }, CodeRecordChanged},
203 "class changed": {func(fx *fixture) { r := fx.record(); r.Hibernate = &no; fx.putRecord(r) }, CodeRecordChanged},
204 }
205 for _, stage := range []string{"before the pick", "planning", "download", "after the last byte"} {
206 for name, c := range changes {
207 t.Run(stage+"/"+name, func(t *testing.T) {
208 t.Parallel()
209 fx := newFixture(t, 2)
210 plan, err := dbdoor.PlanFrom(context.Background(), lsfile.NewReplicaClient(fx.replica), time.Time{})
211 if err != nil {
212 t.Fatal(err)
213 }
214 last := plan.Planned[len(plan.Planned)-1]
215 hook := &stageHook{ReplicaReader: lsfile.NewReplicaClient(fx.replica), stage: stage,
216 lastSpan: dbdoor.Span(last.MinTXID, last.MaxTXID), change: func() { c.change(fx) }}
217 a, _ := fx.agent("uk-1", func(cfg *AgentConfig, _ *inProcWorkers) {
218 cfg.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return hook, nil }
219 })
220 req := fx.request("SELECT count(*) FROM t", "")
221 if stage == "before the pick" {
222 c.change(fx)
223 }
224 _, _, r := serveOne(t, a, fx, req)
225 if r.State != StateRefused || r.Code != c.want {
226 t.Fatalf("result %+v, want refused %s", r, c.want)
227 }
228 if keys := fx.chunkKeys(req.ID); len(keys) != 0 {
229 t.Fatalf("chunks written: %v", keys)
230 }
231 })
232 }
233 }
234}
235
236func TestReadTooLargeAndNoRoom(t *testing.T) {
237 t.Parallel()
238 fx := newFixture(t, 2)
239 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxRestored = 1024 })
240 _, s, r := serveOne(t, a, fx, fx.request("SELECT 1", ""))
241 if r.Code != CodeTooLarge || s.Cap != 1024 || s.Restored <= 1024 {
242 t.Fatalf("too large: %+v %+v", r, s)
243 }
244 fx2 := newFixture(t, 2)
245 b, _ := fx2.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
246 calls := 0
247 c.Free = func() (uint64, uint64, error) {
248 calls++
249 if calls == 1 { // the floor fits; the plan's growth does not
250 return 11 << 30, 20 << 30, nil
251 }
252 return 5<<30 + 1, 20 << 30, nil
253 }
254 })
255 _, _, r2 := serveOne(t, b, fx2, fx2.request("SELECT 1", ""))
256 if r2.Code != CodeNoRoom {
257 t.Fatalf("no room: %+v", r2)
258 }
259}
260
261// Review Focus 4: the cap is exact, a row bigger than a chunk is split, and
262// the summary covers exactly the kept bytes.
263func TestChunkWriterCapIsExact(t *testing.T) {
264 t.Parallel()
265 for _, extra := range []int{0, 1} {
266 fx := newFixture(t, 0)
267 max := int64(3*ChunkBytes + 7)
268 cw := newChunkWriter(context.Background(), fx.store, "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa", fx.reply.PublicKey(), StreamRows, time.Hour, max, ChunkBytes)
269 big := bytes.Repeat([]byte("x"), 2*ChunkBytes+5)
270 rest := bytes.Repeat([]byte("y"), int(max)-len(big)+extra)
271 cw.Write(big)
272 _, err := cw.Write(rest)
273 if (extra == 1) != (err == errCapped) || cw.capped() != (extra == 1) {
274 t.Fatalf("extra %d: Write = %v, capped %v", extra, err, cw.capped())
275 }
276 last, n, sum, err := cw.close()
277 if err != nil || n != max {
278 t.Fatalf("extra %d: kept %d of %d: %v", extra, n, max, err)
279 }
280 want := sha256.Sum256(append(append([]byte{}, big...), rest[:int(max)-len(big)]...))
281 if sum != hex.EncodeToString(want[:]) {
282 t.Fatal("the summary hash is not the kept bytes'")
283 }
284 for seq := int64(1); seq <= last; seq++ {
285 sealed, _, err := fx.store.Get(context.Background(), ChunkKey("dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa", seq))
286 if err != nil || len(sealed)-SealOverhead > ChunkBytes {
287 t.Fatalf("chunk %d: %d bytes, %v", seq, len(sealed), err)
288 }
289 }
290 }
291}
292
293// The output cap end to end: done, output-cap, and exactly the cap kept.
294func TestReadOutputCap(t *testing.T) {
295 t.Parallel()
296 fx := newFixture(t, 2)
297 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxOutput = 100 })
298 rows, s, r := serveOne(t, a, fx, fx.request("SELECT body FROM t", ""))
299 if r.State != StateDone || r.Code != CodeOutputCap || len(rows) != 100 || s.Bytes != 100 {
300 t.Fatalf("result %+v, %d bytes", r, len(rows))
301 }
302}
303
304func TestCancelStopsTheOwner(t *testing.T) {
305 t.Parallel()
306 fx := newFixture(t, 2)
307 hang := make(chan struct{})
308 t.Cleanup(func() { close(hang) })
309 a, _ := fx.agent("uk-1", func(c *AgentConfig, w *inProcWorkers) {
310 c.CancelEvery = 20 * time.Millisecond
311 w.opt.Restore = func(ctx context.Context, _ litestream.ReplicaClient, _ string, _ uint64) error {
312 select {
313 case <-ctx.Done():
314 case <-hang:
315 }
316 return ctx.Err()
317 }
318 })
319 req := fx.request("SELECT 1", "")
320 e := fx.submit(req)
321 a.consider(context.Background(), e)
322 if _, err := fx.store.Put(context.Background(), CancelKey(req.ID), nil, ""); err != nil {
323 t.Fatal(err)
324 }
325 done := make(chan struct{})
326 go func() { a.Wait(); close(done) }()
327 select {
328 case <-done:
329 case <-time.After(5 * time.Second):
330 t.Fatal("the owner did not stop within 5 s of cancel")
331 }
332 if _, _, r := fx.answer(req.ID); r.State != StateCancelled {
333 t.Fatalf("result %+v", r)
334 }
335}
336
337// refusedChunk never lets one chunk's create land: every attempt answers an
338// error and the object stays absent.
339type refusedChunk struct {
340 release.Store
341 key string
342}
343
344func (s refusedChunk) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
345 if key == s.key {
346 return "", errors.New("connection reset")
347 }
348 return s.Store.Put(ctx, key, body, ifMatch)
349}
350
351// Plan review finding 9: a chunk that is never confirmed is never named by
352// the result. The read ends interrupted, and every chunk through its
353// last_seq exists. Mutation: flushLocked increments seq before the create is
354// confirmed; the result then names chunk 2 and this test is red.
355func TestResultNamesOnlyConfirmedChunks(t *testing.T) {
356 t.Parallel()
357 fx := newFixture(t, 1)
358 req := fx.request("SELECT hex(randomblob(300000))", "") // about 600 KB: three chunks
359 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
360 c.Store = refusedChunk{Store: fx.store, key: ChunkKey(req.ID, 2)}
361 c.OwnerDeadline = 2 * time.Second
362 })
363 a.consider(context.Background(), fx.submit(req))
364 a.Wait()
365 raw, _, err := fx.store.Get(context.Background(), ResultKey(req.ID))
366 if err != nil {
367 t.Fatal(err)
368 }
369 var r Result
370 json.Unmarshal(raw, &r)
371 if r.State != StateFailed || r.Code != CodeInterrupted || r.LastSeq != 1 {
372 t.Fatalf("result %+v", r)
373 }
374 for seq := int64(1); seq <= r.LastSeq; seq++ {
375 if _, _, err := fx.store.Get(context.Background(), ChunkKey(req.ID, seq)); err != nil {
376 t.Fatalf("the result names chunk %d, which does not exist", seq)
377 }
378 }
379}
380
381// strictStore fails the test on any protocol delete or replacing write
382// under db-reads/ (T16, the box's half).
383type strictStore struct {
384 release.Store
385 t *testing.T
386}
387
388func (s strictStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
389 if strings.HasPrefix(key, Prefix) && ifMatch != "" {
390 s.t.Errorf("a replacing write of %s", key)
391 }
392 return s.Store.Put(ctx, key, body, ifMatch)
393}
394
395func (s strictStore) Delete(ctx context.Context, key string) error {
396 if strings.HasPrefix(key, Prefix) {
397 s.t.Errorf("a delete of %s", key)
398 }
399 return s.Store.Delete(ctx, key)
400}
401
402func TestReadNeverDeletesOrReplaces(t *testing.T) {
403 t.Parallel()
404 fx := newFixture(t, 2)
405 a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.Store = strictStore{Store: fx.store, t: t} })
406 if _, _, r := serveOne(t, a, fx, fx.request("SELECT count(*) FROM t", "")); r.State != StateDone {
407 t.Fatalf("%+v", r)
408 }
409}
410
411func TestOwnerDeadlineEndsAHungRead(t *testing.T) {
412 t.Parallel()
413 fx := newFixture(t, 1)
414 var started atomic.Bool
415 a, _ := fx.agent("uk-1", func(c *AgentConfig, w *inProcWorkers) {
416 c.OwnerDeadline = 200 * time.Millisecond
417 w.opt.Restore = func(ctx context.Context, _ litestream.ReplicaClient, _ string, _ uint64) error {
418 started.Store(true)
419 <-ctx.Done()
420 return ctx.Err()
421 }
422 })
423 _, _, r := serveOne(t, a, fx, fx.request("SELECT 1", ""))
424 if !started.Load() || r.State != StateFailed || r.Code != CodeInterrupted {
425 t.Fatalf("result %+v", r)
426 }
427}
428
429// Astra round 1, finding 2 (T9 extended): every replica-derived ending (a
430// failed restore's error text, the restored size, the oldest point kept,
431// "nothing replicated", "no room", "changed twice") waits for check B. When
432// the record changed after check A, the read ends refused record-changed
433// and its sealed summary carries none of that replica's details.
434func TestReplicaDetailWaitsForCheckB(t *testing.T) {
435 t.Parallel()
436 recreate := func(fx *fixture) { r := fx.record(); r.CreatedAt = r.CreatedAt.Add(time.Hour); fx.putRecord(r) }
437 for name, c := range map[string]struct {
438 batches int
439 at string
440 vanish int
441 mut func(*AgentConfig, *inProcWorkers)
442 }{
443 "a failing restore": {batches: 2, mut: func(_ *AgentConfig, w *inProcWorkers) {
444 w.opt.Restore = func(context.Context, litestream.ReplicaClient, string, uint64) error {
445 return errors.New("malformed database schema (secret_customer_ledger)")
446 }
447 }},
448 "too large": {batches: 2, mut: func(c *AgentConfig, _ *inProcWorkers) { c.MaxRestored = 1024 }},
449 "before the history": {batches: 2, at: "2000-01-01T00:00:00Z"},
450 "nothing replicated": {batches: 0},
451 "no room": {batches: 2, mut: func(c *AgentConfig, _ *inProcWorkers) {
452 calls := 0
453 c.Free = func() (uint64, uint64, error) {
454 calls++
455 if calls == 1 {
456 return 11 << 30, 20 << 30, nil
457 }
458 return 5<<30 + 1, 20 << 30, nil
459 }
460 }},
461 "changed twice": {batches: 2, vanish: 2},
462 } {
463 t.Run(name, func(t *testing.T) {
464 t.Parallel()
465 fx := newFixture(t, c.batches)
466 var rd dbdoor.ReplicaReader = &stageHook{ReplicaReader: lsfile.NewReplicaClient(fx.replica), stage: "planning", change: func() { recreate(fx) }}
467 if c.vanish > 0 {
468 // The record changes during the second attempt's download,
469 // after that attempt's check A: only a check before
470 // "changed twice" can see it.
471 rd = &vanishThenChange{ReplicaReader: &vanishing{ReplicaReader: lsfile.NewReplicaClient(fx.replica), times: c.vanish}, at: 2, change: func() { recreate(fx) }}
472 }
473 a, _ := fx.agent("uk-1", func(cfg *AgentConfig, w *inProcWorkers) {
474 cfg.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return rd, nil }
475 if c.mut != nil {
476 c.mut(cfg, w)
477 }
478 })
479 _, s, r := serveOne(t, a, fx, fx.request("SELECT 1", c.at))
480 if r.State != StateRefused || r.Code != CodeRecordChanged {
481 t.Fatalf("result %+v, want refused %s", r, CodeRecordChanged)
482 }
483 if !reflect.DeepEqual(s, Summary{}) {
484 t.Fatalf("the refused read's summary carries the replica's details: %+v", s)
485 }
486 })
487 }
488}
489
490// vanishThenChange runs change on the at'th whole-file open, before the
491// open answers; over vanishing, that open is one that vanishes.
492type vanishThenChange struct {
493 dbdoor.ReplicaReader
494 mu sync.Mutex
495 n, at int
496 change func()
497}
498
499func (v *vanishThenChange) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, off, size int64) (io.ReadCloser, error) {
500 if off == 0 && size == 0 {
501 v.mu.Lock()
502 v.n++
503 fire := v.n == v.at
504 v.mu.Unlock()
505 if fire {
506 v.change()
507 }
508 }
509 return v.ReplicaReader.OpenLTXFile(ctx, level, lo, hi, off, size)
510}
511
512// slowPicked lands the picked create and then answers late: an election
513// whose response takes delay (Astra round 1, finding 3).
514type slowPicked struct {
515 release.Store
516 delay time.Duration
517}
518
519func (s slowPicked) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
520 etag, err := s.Store.Put(ctx, key, body, ifMatch)
521 if strings.HasSuffix(key, "/picked") {
522 time.Sleep(s.delay)
523 }
524 return etag, err
525}
526
527// The owner works to the deadline it published in picked, not to a fresh
528// one from whenever the election's answer arrived (Astra round 1, finding
529// 3): a hung read ends by picked.Deadline, and a read whose election
530// answered after its deadline starts no work at all.
531func TestServeKeepsThePublishedDeadline(t *testing.T) {
532 t.Parallel()
533 for name, c := range map[string]struct {
534 deadline, delay time.Duration
535 dials int64
536 }{
537 "a late answer inside the deadline": {deadline: 600 * time.Millisecond, delay: 300 * time.Millisecond, dials: 1},
538 "an answer after the deadline": {deadline: 200 * time.Millisecond, delay: 300 * time.Millisecond, dials: 0},
539 } {
540 t.Run(name, func(t *testing.T) {
541 t.Parallel()
542 fx := newFixture(t, 1)
543 a, w := fx.agent("uk-1", func(cfg *AgentConfig, w *inProcWorkers) {
544 cfg.Store = slowPicked{Store: fx.store, delay: c.delay}
545 cfg.OwnerDeadline = c.deadline
546 w.opt.Restore = func(ctx context.Context, _ litestream.ReplicaClient, _ string, _ uint64) error {
547 <-ctx.Done()
548 return ctx.Err()
549 }
550 })
551 req := fx.request("SELECT 1", "")
552 _, _, r := serveOne(t, a, fx, req)
553 raw, _, err := fx.store.Get(context.Background(), PickedKey(req.ID))
554 if err != nil {
555 t.Fatal(err)
556 }
557 var p Picked
558 if err := json.Unmarshal(raw, &p); err != nil {
559 t.Fatal(err)
560 }
561 if r.State != StateFailed || r.Code != CodeInterrupted || w.dials.Load() != c.dials {
562 t.Fatalf("result %+v, %d worker connections", r, w.dials.Load())
563 }
564 if over := r.Ended.Sub(p.Deadline); over > 150*time.Millisecond {
565 t.Fatalf("the read ran %s past the deadline it published", over)
566 }
567 })
568 }
569}
570