1package work23import (4 "context"5 "database/sql"6 "errors"7 "sync/atomic"8 "testing"9 "time"1011 "github.com/FelineStateMachine/tinyrelay/internal/storage"12 "github.com/prometheus/client_golang/prometheus"13)1415func 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}4748func 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}6061func 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 bool70 if err := store.WithTx(ctx, func(tx *sql.Tx) error {71 var err error72 completed, err = CompletePendingTx(ctx, tx, pendingID, time.Now())73 return err74 }); err != nil {75 t.Fatal(err)76 }77 if !completed {78 t.Fatal("pending intent was not completed")79 }8081 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 err93 }94 if ok {95 t.Fatal("running intent was completed")96 }97 return nil98 }); 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}106107func 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}131132func 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 break158 }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}169170func 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}191192func 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}217218func 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.Int32227 var maximum atomic.Int32228 var calls atomic.Int32229 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 break236 }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}271272func 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}291292func 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}325326func 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}361362func 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 nil381 }}, 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}405406func 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 continue423 }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}433434func 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 store442}