tinyrelayfile
tinyrelay/internal/work/work_test.go/
internal/work/work_test.go raw
1package work
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/FelineStateMachine/tinyrelay/internal/storage"
12 "github.com/prometheus/client_golang/prometheus"
13)
14
15func TestQueueClaimsFencesAndRetries(t *testing.T) {
16 ctx := context.Background()
17 store := openStore(t)
18 queue := New(store)
19 now := time.Unix(100, 0)
20 id, err := queue.Enqueue(ctx, Intent{Kind: "delivery", EventID: "event-1", Target: "target", Payload: "{}", NextAt: now})
21 if err != nil {
22 t.Fatal(err)
23 }
24 claimed, err := queue.Claim(ctx, now, time.Minute)
25 if err != nil || claimed == nil {
26 t.Fatalf("claim = %#v, %v", claimed, err)
27 }
28 if claimed.ID != id || claimed.Attempts != 1 || claimed.ClaimToken == "" {
29 t.Fatalf("bad claim: %#v", claimed)
30 }
31 ok, err := queue.Complete(ctx, claimed.ID, "wrong", now)
32 if err != nil || ok {
33 t.Fatal("wrong token completed claim")
34 }
35 ok, err = queue.Retry(ctx, claimed.ID, claimed.ClaimToken, errors.New("temporary"), now)
36 if err != nil || !ok {
37 t.Fatalf("retry = %v, %v", ok, err)
38 }
39 items, err := queue.Pending(ctx, "delivery")
40 if err != nil || len(items) != 1 {
41 t.Fatalf("pending = %#v, %v", items, err)
42 }
43 if items[0].LastError != "temporary" || !items[0].NextAt.After(now) {
44 t.Fatalf("retry metadata = %#v", items[0])
45 }
46}
47
48func TestEnqueueUsesSharedIntentID(t *testing.T) {
49 store := openStore(t)
50 queue := New(store)
51 intent := Intent{Kind: "delivery", EventID: "event-1", Target: "target"}
52 id, err := queue.Enqueue(context.Background(), intent)
53 if err != nil {
54 t.Fatal(err)
55 }
56 if want := storage.IntentID(intent.Kind, intent.EventID, intent.Target); id != want {
57 t.Fatalf("intent id = %q, want %q", id, want)
58 }
59}
60
61func TestCompletePendingTxDoesNotTouchRunningIntent(t *testing.T) {
62 store := openStore(t)
63 queue := New(store)
64 ctx := context.Background()
65 pendingID, err := queue.Enqueue(ctx, Intent{Kind: "pending", EventID: "pending", Target: "target"})
66 if err != nil {
67 t.Fatal(err)
68 }
69 var completed bool
70 if err := store.WithTx(ctx, func(tx *sql.Tx) error {
71 var err error
72 completed, err = CompletePendingTx(ctx, tx, pendingID, time.Now())
73 return err
74 }); err != nil {
75 t.Fatal(err)
76 }
77 if !completed {
78 t.Fatal("pending intent was not completed")
79 }
80
81 runningID, err := queue.Enqueue(ctx, Intent{Kind: "running", EventID: "running", Target: "target"})
82 if err != nil {
83 t.Fatal(err)
84 }
85 claimed, err := queue.Claim(ctx, time.Now(), time.Minute)
86 if err != nil || claimed == nil || claimed.ID != runningID {
87 t.Fatalf("claim = %#v, %v", claimed, err)
88 }
89 if err := store.WithTx(ctx, func(tx *sql.Tx) error {
90 ok, err := CompletePendingTx(ctx, tx, runningID, time.Now())
91 if err != nil {
92 return err
93 }
94 if ok {
95 t.Fatal("running intent was completed")
96 }
97 return nil
98 }); err != nil {
99 t.Fatal(err)
100 }
101 items, err := queue.List(ctx, "running")
102 if err != nil || len(items) != 1 {
103 t.Fatalf("running intents = %#v, %v", items, err)
104 }
105}
106
107func TestCancelClaimRequiresCurrentToken(t *testing.T) {
108 store := openStore(t)
109 queue := New(store)
110 ctx := context.Background()
111 id, err := queue.Enqueue(ctx, Intent{Kind: "terminal", EventID: "event", Target: "target"})
112 if err != nil {
113 t.Fatal(err)
114 }
115 claimed, err := queue.Claim(ctx, time.Now(), time.Minute)
116 if err != nil || claimed == nil || claimed.ID != id {
117 t.Fatalf("claim = %#v, %v", claimed, err)
118 }
119 ok, err := queue.CancelClaim(ctx, id, "stale-token", errors.New("stale"), time.Now())
120 if err != nil || ok {
121 t.Fatalf("stale cancel = %v, %v", ok, err)
122 }
123 if ok, err := queue.CancelClaim(ctx, id, claimed.ClaimToken, errors.New("terminal"), time.Now()); err != nil || !ok {
124 t.Fatalf("current cancel = %v, %v", ok, err)
125 }
126 items, err := queue.List(ctx, "cancelled")
127 if err != nil || len(items) != 1 || items[0].LastError != "terminal" {
128 t.Fatalf("cancelled intents = %#v, %v", items, err)
129 }
130}
131
132func TestWorkerStopCancelsClaimWithoutRetry(t *testing.T) {
133 store := openStore(t)
134 queue := New(store)
135 ctx := context.Background()
136 if _, err := queue.Enqueue(ctx, Intent{Kind: "terminal", EventID: "event", Target: "target"}); err != nil {
137 t.Fatal(err)
138 }
139 workerCtx, cancel := context.WithCancel(ctx)
140 defer cancel()
141 done := make(chan error, 1)
142 go func() {
143 done <- NewWorker(queue, map[string]Handler{"terminal": func(context.Context, Intent) error {
144 return Stop(errors.New("permanent"))
145 }}).Run(workerCtx)
146 }()
147 deadline := time.Now().Add(time.Second)
148 for {
149 items, err := queue.List(ctx, "cancelled")
150 if err != nil {
151 t.Fatal(err)
152 }
153 if len(items) == 1 {
154 if items[0].LastError != "permanent" {
155 t.Fatalf("last error = %q", items[0].LastError)
156 }
157 break
158 }
159 if time.Now().After(deadline) {
160 t.Fatal("worker did not cancel terminal intent")
161 }
162 time.Sleep(time.Millisecond)
163 }
164 cancel()
165 if err := <-done; !errors.Is(err, context.Canceled) {
166 t.Fatalf("worker error = %v", err)
167 }
168}
169
170func TestExpiredClaimCanBeReclaimed(t *testing.T) {
171 ctx := context.Background()
172 store := openStore(t)
173 queue := New(store)
174 now := time.Unix(100, 0)
175 if _, err := queue.Enqueue(ctx, Intent{Kind: "push", EventID: "event-1", Target: "target", NextAt: now}); err != nil {
176 t.Fatal(err)
177 }
178 one, err := queue.Claim(ctx, now, time.Second)
179 if err != nil {
180 t.Fatal(err)
181 }
182 two, err := queue.Claim(ctx, now.Add(2*time.Second), time.Second)
183 if err != nil || two == nil || two.ClaimToken == one.ClaimToken {
184 t.Fatalf("reclaim = %#v, %v", two, err)
185 }
186 ok, err := queue.Complete(ctx, one.ID, one.ClaimToken, now.Add(2*time.Second))
187 if err != nil || ok {
188 t.Fatal("expired token completed reclaimed work")
189 }
190}
191
192func TestWorkerLeavesClaimReclaimableOnCancellation(t *testing.T) {
193 ctx, cancel := context.WithCancel(context.Background())
194 store := openStore(t)
195 queue := New(store)
196 if _, err := queue.Enqueue(context.Background(), Intent{Kind: "slow", EventID: "event-1", Target: "target"}); err != nil {
197 t.Fatal(err)
198 }
199 started := make(chan struct{})
200 worker := NewWorker(queue, map[string]Handler{"slow": func(context.Context, Intent) error { close(started); <-ctx.Done(); return ctx.Err() }})
201 errCh := make(chan error, 1)
202 go func() { errCh <- worker.Run(ctx) }()
203 select {
204 case <-started:
205 case <-time.After(time.Second):
206 t.Fatal("worker did not start")
207 }
208 cancel()
209 if err := <-errCh; !errors.Is(err, context.Canceled) {
210 t.Fatalf("worker error = %v", err)
211 }
212 items, err := queue.List(context.Background(), "running")
213 if err != nil || len(items) != 1 {
214 t.Fatalf("running = %#v, %v", items, err)
215 }
216}
217
218func TestWorkersRenewLongAttemptWithoutConcurrentDuplicate(t *testing.T) {
219 store := openStore(t)
220 queue := New(store)
221 if _, err := queue.Enqueue(context.Background(), Intent{Kind: "slow", EventID: "event-1", Target: "target"}); err != nil {
222 t.Fatal(err)
223 }
224 ctx, cancel := context.WithCancel(context.Background())
225 defer cancel()
226 var active atomic.Int32
227 var maximum atomic.Int32
228 var calls atomic.Int32
229 handler := func(handlerCtx context.Context, _ Intent) error {
230 calls.Add(1)
231 current := active.Add(1)
232 for {
233 previous := maximum.Load()
234 if current <= previous || maximum.CompareAndSwap(previous, current) {
235 break
236 }
237 }
238 defer active.Add(-1)
239 <-handlerCtx.Done()
240 return handlerCtx.Err()
241 }
242 workerOptions := WorkerOptions{Lease: 2 * time.Second, Renew: 500 * time.Millisecond}
243 workers := []*Worker{NewWorkerWithOptions(queue, map[string]Handler{"slow": handler}, workerOptions), NewWorkerWithOptions(queue, map[string]Handler{"slow": handler}, workerOptions)}
244 errs := make(chan error, len(workers))
245 for _, worker := range workers {
246 go func() { errs <- worker.Run(ctx) }()
247 }
248 deadline := time.After(4 * time.Second)
249 for calls.Load() == 0 {
250 select {
251 case <-deadline:
252 t.Fatal("worker did not start")
253 default:
254 time.Sleep(10 * time.Millisecond)
255 }
256 }
257 time.Sleep(2500 * time.Millisecond)
258 if got := maximum.Load(); got != 1 {
259 t.Fatalf("maximum concurrent handlers = %d", got)
260 }
261 cancel()
262 for range workers {
263 if err := <-errs; !errors.Is(err, context.Canceled) {
264 t.Fatalf("worker error = %v", err)
265 }
266 }
267 if calls.Load() != 1 {
268 t.Fatalf("handler calls = %d", calls.Load())
269 }
270}
271
272func TestRunPoolCancelsWorkersOnQueueError(t *testing.T) {
273 store := openStore(t)
274 queue := New(store)
275 done := make(chan error, 1)
276 go func() {
277 done <- RunPool(context.Background(), queue, map[string]Handler{}, PoolOptions{Workers: 2})
278 }()
279 if err := store.Close(); err != nil {
280 t.Fatal(err)
281 }
282 select {
283 case err := <-done:
284 if err == nil {
285 t.Fatal("RunPool returned nil after queue failure")
286 }
287 case <-time.After(time.Second):
288 t.Fatal("RunPool did not stop after queue failure")
289 }
290}
291
292func TestRunPoolWaitsForActiveHandlersOnCancellation(t *testing.T) {
293 store := openStore(t)
294 queue := New(store)
295 if _, err := queue.Enqueue(context.Background(), Intent{Kind: "slow", EventID: "event-1", Target: "target"}); err != nil {
296 t.Fatal(err)
297 }
298 ctx, cancel := context.WithCancel(context.Background())
299 started := make(chan struct{})
300 done := make(chan error, 1)
301 go func() {
302 done <- RunPool(ctx, queue, map[string]Handler{
303 "slow": func(handlerCtx context.Context, _ Intent) error {
304 close(started)
305 <-handlerCtx.Done()
306 return handlerCtx.Err()
307 },
308 }, PoolOptions{Workers: 1})
309 }()
310 select {
311 case <-started:
312 case <-time.After(time.Second):
313 t.Fatal("pool did not start handler")
314 }
315 cancel()
316 select {
317 case err := <-done:
318 if !errors.Is(err, context.Canceled) {
319 t.Fatalf("RunPool error = %v", err)
320 }
321 case <-time.After(time.Second):
322 t.Fatal("RunPool did not wait for active handler")
323 }
324}
325
326func TestWorkerCancellationFencesActiveAttempt(t *testing.T) {
327 store := openStore(t)
328 queue := New(store)
329 intentID, err := queue.Enqueue(context.Background(), Intent{Kind: "cancel", EventID: "event-1", Target: "target"})
330 if err != nil {
331 t.Fatal(err)
332 }
333 ctx, cancel := context.WithCancel(context.Background())
334 defer cancel()
335 started := make(chan struct{})
336 worker := NewWorkerWithOptions(queue, map[string]Handler{"cancel": func(handlerCtx context.Context, _ Intent) error {
337 close(started)
338 <-handlerCtx.Done()
339 return handlerCtx.Err()
340 }}, WorkerOptions{Lease: 2 * time.Second, Renew: 100 * time.Millisecond})
341 errCh := make(chan error, 1)
342 go func() { errCh <- worker.Run(ctx) }()
343 select {
344 case <-started:
345 case <-time.After(time.Second):
346 t.Fatal("worker did not start")
347 }
348 if err := queue.Cancel(context.Background(), intentID, time.Now()); err != nil {
349 t.Fatal(err)
350 }
351 time.Sleep(300 * time.Millisecond)
352 cancel()
353 if err := <-errCh; !errors.Is(err, context.Canceled) {
354 t.Fatalf("worker error = %v", err)
355 }
356 items, err := queue.List(context.Background(), "cancelled")
357 if err != nil || len(items) != 1 {
358 t.Fatalf("cancelled = %#v, %v", items, err)
359 }
360}
361
362func TestWorkerContinuesAfterCancellationFencesAttempt(t *testing.T) {
363 store := openStore(t)
364 queue := New(store)
365 firstID, err := queue.Enqueue(context.Background(), Intent{Kind: "cancel", EventID: "first", Target: "target"})
366 if err != nil {
367 t.Fatal(err)
368 }
369 ctx, cancel := context.WithCancel(context.Background())
370 defer cancel()
371 started := make(chan struct{})
372 secondDone := make(chan struct{})
373 worker := NewWorkerWithOptions(queue, map[string]Handler{"cancel": func(handlerCtx context.Context, intent Intent) error {
374 if intent.EventID == "first" {
375 close(started)
376 <-handlerCtx.Done()
377 return handlerCtx.Err()
378 }
379 close(secondDone)
380 return nil
381 }}, WorkerOptions{Lease: 2 * time.Second, Renew: 50 * time.Millisecond})
382 errCh := make(chan error, 1)
383 go func() { errCh <- worker.Run(ctx) }()
384 select {
385 case <-started:
386 case <-time.After(time.Second):
387 t.Fatal("worker did not start first intent")
388 }
389 if err := queue.Cancel(context.Background(), firstID, time.Now()); err != nil {
390 t.Fatal(err)
391 }
392 if _, err := queue.Enqueue(context.Background(), Intent{Kind: "cancel", EventID: "second", Target: "target"}); err != nil {
393 t.Fatal(err)
394 }
395 select {
396 case <-secondDone:
397 case <-time.After(2 * time.Second):
398 t.Fatal("worker stopped after fenced cancellation")
399 }
400 cancel()
401 if err := <-errCh; !errors.Is(err, context.Canceled) {
402 t.Fatalf("worker error = %v", err)
403 }
404}
405
406func TestQueueMetricsNormalizeUnknownKinds(t *testing.T) {
407 store := openStore(t)
408 queue := New(store)
409 registry := prometheus.NewPedanticRegistry()
410 if err := registry.Register(queue.Metrics()); err != nil {
411 t.Fatal(err)
412 }
413 if _, err := queue.Enqueue(context.Background(), Intent{Kind: "tenant-supplied-kind", EventID: "event-1", Target: "target"}); err != nil {
414 t.Fatal(err)
415 }
416 families, err := registry.Gather()
417 if err != nil {
418 t.Fatal(err)
419 }
420 for _, family := range families {
421 if family.GetName() != "relay_work_pending" {
422 continue
423 }
424 for _, metric := range family.GetMetric() {
425 for _, label := range metric.GetLabel() {
426 if label.GetName() == "kind" && label.GetValue() == "tenant-supplied-kind" {
427 t.Fatal("unbounded work kind label")
428 }
429 }
430 }
431 }
432}
433
434func openStore(t *testing.T) *storage.Store {
435 t.Helper()
436 store, err := storage.Open(context.Background(), t.TempDir()+"/relay.db")
437 if err != nil {
438 t.Fatal(err)
439 }
440 t.Cleanup(func() { _ = store.Close() })
441 return store
442}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs