From 208c69c4336b32ebe5b602cc858083b562e2598f Mon Sep 17 00:00:00 2001 From: opencode Date: Sat, 3 Oct 2026 21:27:04 +0800 Subject: [PATCH] test: add queue receive timeout integration coverage (framework#1568) --- config/queue.go | 3 + go.mod | 6 +- go.sum | 8 +- tests/feature/queue_test.go | 406 ++++++++++++++++++++++++++++++++++++ 4 files changed, 416 insertions(+), 7 deletions(-) diff --git a/config/queue.go b/config/queue.go index 788c5eb..6e11c9a 100644 --- a/config/queue.go +++ b/config/queue.go @@ -26,6 +26,9 @@ func init() { "connection": "sqlite", "queue": "default", "concurrent": 5, + // timeout is the number of seconds to wait for a blocking receive, e.g. 5 or "1s". + // Only affects receive-based (DriverWithReceive) connections. + "timeout": 5, // Retry_after is the number of seconds a job is reserved for before a // crashed worker's reservation expires and the job is recovered by // other workers. It must exceed the maximum job runtime. diff --git a/go.mod b/go.mod index 5b960e6..fbf3942 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/goravel/cos v1.18.0 github.com/goravel/example-proto v0.0.1 github.com/goravel/fiber v1.18.1-0.20260901101548-72e06e6c5635 - github.com/goravel/framework v1.18.1-0.20260925125002-93850001fe6f + github.com/goravel/framework v1.18.1-0.20261003042419-11d7b3085e4a github.com/goravel/gemini v1.18.0 github.com/goravel/gin v1.18.1-0.20260926005436-467f03c48058 github.com/goravel/inertia v1.18.1-0.20260824090129-ac3d472d028c @@ -34,7 +34,7 @@ require ( go.opentelemetry.io/otel v1.46.0 go.opentelemetry.io/otel/metric v1.46.0 go.opentelemetry.io/otel/trace v1.46.0 - google.golang.org/grpc v1.85.0-dev.0.20260825072537-93e31b48545e + google.golang.org/grpc v1.86.0-dev.0.20260925073600-acccf8cd101a google.golang.org/protobuf v1.36.12 ) @@ -264,4 +264,4 @@ require ( gorm.io/plugin/dbresolver v1.6.2 // indirect ) -replace github.com/goravel/framework => github.com/goravel/framework v1.18.1-0.20260925125002-93850001fe6f +replace github.com/goravel/framework => github.com/goravel/framework v1.18.1-0.20261003042419-11d7b3085e4a diff --git a/go.sum b/go.sum index 57c4521..3c40950 100644 --- a/go.sum +++ b/go.sum @@ -317,8 +317,8 @@ github.com/goravel/example-proto v0.0.1 h1:ZxETeKREQWjuJ49bX/Hqj1NLR5Vyj489Ks6dR github.com/goravel/example-proto v0.0.1/go.mod h1:I8IPsHr4Ndf7KxmdsRpBR2LQ0Geo48+pjv9IIWf3mZg= github.com/goravel/fiber v1.18.1-0.20260901101548-72e06e6c5635 h1:iqzZ0P+KC8uBFNQ7PISL81U8CUc9myzS14NRYfftOB4= github.com/goravel/fiber v1.18.1-0.20260901101548-72e06e6c5635/go.mod h1:fk0YaivBle83SlSsPziy5+TRMavqlAr/QeVNo6pRmJA= -github.com/goravel/framework v1.18.1-0.20260925125002-93850001fe6f h1:/hWkvKBhtOPb6k1awR5scXcaLTHS+r1Rh3pS9O0DVIg= -github.com/goravel/framework v1.18.1-0.20260925125002-93850001fe6f/go.mod h1:7Phix1gVUYzA1muoX1o92AJszuDgBUgO31ICCCJ7+Xw= +github.com/goravel/framework v1.18.1-0.20261003042419-11d7b3085e4a h1:rImd+l3pAotzOtGXQuMEnqE7gby/l0e8KeG0Bm4Ephs= +github.com/goravel/framework v1.18.1-0.20261003042419-11d7b3085e4a/go.mod h1:ccePbn4R6XOv3eoZGR2Kg4c7ZQlyDVdR5V+T3IH0mJw= github.com/goravel/gemini v1.18.0 h1:Tv6IJ3KvBvH5QD7kAM3XFQR0g8IUfkqMbRGi7dtlWUo= github.com/goravel/gemini v1.18.0/go.mod h1:b4FueB0O0Qkz71eVWLiSaKghp+Eb/1hNNICkk7vMgrQ= github.com/goravel/gin v1.18.1-0.20260926005436-467f03c48058 h1:EId8wBSolUZQ00QQz4fyPrRWN+Tz/TZAw4u65XwSINY= @@ -841,8 +841,8 @@ google.golang.org/genproto/googleapis/api v0.0.0-20260819154853-08b0e4226688 h1: google.golang.org/genproto/googleapis/api v0.0.0-20260819154853-08b0e4226688/go.mod h1:1RJ9BQGyNdZwkGc1eTqkErfRZ6RJyYPHZo73BZ1vQqI= google.golang.org/genproto/googleapis/rpc v0.0.0-20260825221802-da73d73af1c5 h1:1VUiZAXyC+zmiFYi+WLtBzr68Cj8wOofHjjrA/kkizc= google.golang.org/genproto/googleapis/rpc v0.0.0-20260825221802-da73d73af1c5/go.mod h1:DjtHYE8FKJLivXcBEjGwndXfIC23G0VpXiXKqG179uA= -google.golang.org/grpc v1.85.0-dev.0.20260825072537-93e31b48545e h1:ar+feAPij1Znzug45m6AoShUCc5AaR8FLA3I6LZd94k= -google.golang.org/grpc v1.85.0-dev.0.20260825072537-93e31b48545e/go.mod h1:7aY/KlAHanMAVIWHnk8wloQsdlfwzDz/PP9vCbJG6CU= +google.golang.org/grpc v1.86.0-dev.0.20260925073600-acccf8cd101a h1:QDqoSLTVtcul9fIYy+E/WsMzRtpuKCaQS2zxqZpDGq4= +google.golang.org/grpc v1.86.0-dev.0.20260925073600-acccf8cd101a/go.mod h1:2CcatsWDqg61/O3fuK5USzCqFX+N/4WHBioUxRNhCK0= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/tests/feature/queue_test.go b/tests/feature/queue_test.go index 43bbfd7..fddb41c 100644 --- a/tests/feature/queue_test.go +++ b/tests/feature/queue_test.go @@ -1,7 +1,9 @@ package feature import ( + "context" "errors" + "sync" "testing" "time" @@ -18,12 +20,211 @@ import ( type QueueTestSuite struct { suite.Suite tests.TestCase + + timeoutDriver *receiveTimeoutDriver + savedConnections map[string]any } func TestQueueTestSuite(t *testing.T) { suite.Run(t, &QueueTestSuite{}) } +const timeoutTestConnection = "timeout_receive" + +var ( + _ contractsqueue.Driver = (*receiveTimeoutDriver)(nil) + _ contractsqueue.DriverWithReceive = (*receiveTimeoutDriver)(nil) +) + +// receiveCall records one Receive invocation: its arguments and the deadline of +// the context the worker passed in. +type receiveCall struct { + queue string + count int + entered time.Time + deadline time.Time + hasDeadline bool +} + +// receiveTimeoutDriver implements Driver + DriverWithReceive and records every +// Receive call, so the suite can prove `queue.connections..timeout` +// reaches the blocking receive call and bounds it. +type receiveTimeoutDriver struct { + mu sync.Mutex + receivedCalls []receiveCall + popQueues []string + deliver bool // when true, a non-blocked Receive hands back one job + block bool // when true, the first Receive waits until the context deadline + deliveredJob *timeoutReservedJob + firstReceive chan struct{} + secondReceive chan struct{} + popSignal chan struct{} + firstOnce sync.Once + secondOnce sync.Once + popOnce sync.Once +} + +func newReceiveTimeoutDriver() *receiveTimeoutDriver { + return &receiveTimeoutDriver{ + firstReceive: make(chan struct{}), + secondReceive: make(chan struct{}), + popSignal: make(chan struct{}), + } +} + +func (d *receiveTimeoutDriver) Driver() string { return contractsqueue.DriverCustom } +func (d *receiveTimeoutDriver) Push(contractsqueue.Task, string) error { return nil } +func (d *receiveTimeoutDriver) Pop(queue string) (contractsqueue.ReservedJob, error) { + d.mu.Lock() + d.popQueues = append(d.popQueues, queue) + d.mu.Unlock() + + d.popOnce.Do(func() { close(d.popSignal) }) + return nil, nil +} + +func (d *receiveTimeoutDriver) Receive(ctx context.Context, queue string, count int) ([]contractsqueue.ReservedJob, error) { + deadline, hasDeadline := ctx.Deadline() + + d.mu.Lock() + d.receivedCalls = append(d.receivedCalls, receiveCall{ + queue: queue, + count: count, + entered: time.Now(), + deadline: deadline, + hasDeadline: hasDeadline, + }) + callNumber := len(d.receivedCalls) + // Only the first call blocks; later calls take the delivery path so the test + // can prove a job is still handled after a receive timeout. + block := d.block && callNumber == 1 + deliver := d.deliver && !block + if deliver { + d.deliver = false + d.deliveredJob = &timeoutReservedJob{} + } + job := d.deliveredJob + d.mu.Unlock() + + d.firstOnce.Do(func() { close(d.firstReceive) }) + if callNumber >= 2 { + d.secondOnce.Do(func() { close(d.secondReceive) }) + } + + if block { + <-ctx.Done() + return nil, nil + } + if deliver { + return []contractsqueue.ReservedJob{job}, nil + } + return nil, nil +} + +// calls returns a copy of every recorded Receive invocation. Returning a copy +// lets tests assert on all calls without racing the worker goroutine. +func (d *receiveTimeoutDriver) calls() []receiveCall { + d.mu.Lock() + defer d.mu.Unlock() + return append([]receiveCall(nil), d.receivedCalls...) +} + +// pops returns a copy of every queue name passed to Pop. +func (d *receiveTimeoutDriver) pops() []string { + d.mu.Lock() + defer d.mu.Unlock() + return append([]string(nil), d.popQueues...) +} + +// delivered returns the job the fake handed back, if any. Reading it under the +// driver mutex keeps the pointer access race-free. +func (d *receiveTimeoutDriver) delivered() *timeoutReservedJob { + d.mu.Lock() + defer d.mu.Unlock() + return d.deliveredJob +} + +// timeoutReservedJob is the minimal ReservedJob the fake driver returns to prove +// a job is still processed and acknowledged when a custom timeout is configured. +type timeoutReservedJob struct { + mu sync.Mutex + deleteCount int +} + +func (j *timeoutReservedJob) Attempts() int { return 1 } + +func (j *timeoutReservedJob) Delete() error { + j.mu.Lock() + defer j.mu.Unlock() + j.deleteCount++ + return nil +} + +func (j *timeoutReservedJob) Release(time.Duration) error { return nil } +func (j *timeoutReservedJob) Task() contractsqueue.Task { + return contractsqueue.Task{ + ChainJob: contractsqueue.ChainJob{ + Job: &jobs.Test{}, + Args: []contractsqueue.Arg{{Type: "string", Value: "timeout-job"}}, + }, + } +} + +// deletes returns how many times the reservation was acknowledged. +func (j *timeoutReservedJob) deletes() int { + j.mu.Lock() + defer j.mu.Unlock() + return j.deleteCount +} + +// startWorker runs worker in the background and returns an idempotent stop func +// that shuts it down and waits for Run to return. +// +// Callers must observe a first Receive/Pop before calling stop; stopping before +// the worker has started can race its startup. stop bounds both Shutdown and the +// Run wait with timeouts so a wedged worker is reported instead of hanging. +func (s *QueueTestSuite) startWorker(worker contractsqueue.Worker) (stop func()) { + s.T().Helper() + + errCh := make(chan error, 1) + go func() { errCh <- worker.Run() }() + + var once sync.Once + return func() { + once.Do(func() { + shutdownCh := make(chan error, 1) + go func() { shutdownCh <- worker.Shutdown() }() + + select { + case err := <-shutdownCh: + s.NoError(err, "worker shutdown") + case <-time.After(2 * time.Second): + s.Fail("worker shutdown did not return; worker may still be running") + } + + select { + case err := <-errCh: + s.NoError(err, "worker run") + case <-time.After(2 * time.Second): + s.Fail("worker did not stop") + } + }) + } +} + +// SetupSuite snapshots a copy of the queue connections so TearDownSuite can +// restore them without mutating config-owned state. +func (s *QueueTestSuite) SetupSuite() { + connections, ok := facades.Config().Get("queue.connections").(map[string]any) + s.Require().True(ok, "queue.connections is not a map") + + saved := make(map[string]any, len(connections)) + for name, connection := range connections { + saved[name] = connection + } + s.savedConnections = saved +} + // SetupTest will run before each test in the suite. func (s *QueueTestSuite) SetupTest() { jobs.TestResult = nil @@ -35,6 +236,16 @@ func (s *QueueTestSuite) SetupTest() { func (s *QueueTestSuite) TearDownTest() { } +// TearDownSuite restores the snapshotted queue connections so the test-only +// connection does not leak into later suites. +func (s *QueueTestSuite) TearDownSuite() { + if s.savedConnections == nil { + return + } + + facades.Config().Add("queue.connections", s.savedConnections) +} + func (s *QueueTestSuite) TestDispatch() { s.NoError(facades.Queue().Job(&jobs.Test{}, testQueueArgs).Dispatch()) @@ -101,6 +312,12 @@ func (s *QueueTestSuite) TestSyncFailedJob() { s.Equal(errors.New("test error"), facades.Queue().Job(&jobs.TestErr{}).Dispatch()) } +// TestDatabaseConnectionTimeoutScaffold asserts the documented scaffold value on +// the database connection, which the timeout tests otherwise never read. +func (s *QueueTestSuite) TestDatabaseConnectionTimeoutScaffold() { + s.Equal(5, facades.Config().GetInt("queue.connections.database.timeout")) +} + func (s *QueueTestSuite) TestFailedJobAndRetry() { if facades.Config().GetString("queue.default") == "sync" { s.T().Skip("skip test due to only for non-sync") @@ -210,6 +427,195 @@ func (s *QueueTestSuite) TestReleaseBasedRetryExhausted() { s.Equal([]any{"exhausted", "exhausted", "exhausted"}, jobs.TestRetryableResult) } +func (s *QueueTestSuite) TestConfiguredTimeoutBoundsReceive() { + cases := []struct { + name string + include bool + value any + expect time.Duration + }{ + {name: "integer seconds", include: true, value: 10, expect: 10 * time.Second}, + {name: "fractional seconds", include: true, value: 2.5, expect: 2500 * time.Millisecond}, + {name: "numeric string", include: true, value: "3", expect: 3 * time.Second}, + {name: "duration string", include: true, value: "1500ms", expect: 1500 * time.Millisecond}, + {name: "missing key falls back to default", include: false, value: nil, expect: contractsqueue.DefaultReceiveTimeout}, + {name: "zero falls back to default", include: true, value: 0, expect: contractsqueue.DefaultReceiveTimeout}, + {name: "negative falls back to default", include: true, value: -3, expect: contractsqueue.DefaultReceiveTimeout}, + {name: "unparsable string falls back to default", include: true, value: "not-a-duration", expect: contractsqueue.DefaultReceiveTimeout}, + } + + for _, tt := range cases { + s.Run(tt.name, func() { + // testify does not run SetupTest between subtests; allocate a fresh + // driver and vary the connection map directly. + s.timeoutDriver = newReceiveTimeoutDriver() + s.registerTimeoutConnection(tt.include, tt.value) + + s.runTimeoutWorker() + + calls := s.timeoutDriver.calls() + s.Require().NotEmpty(calls, "Receive should have been called") + s.Equal("critical", calls[0].queue) + s.Equal(3, calls[0].count) + s.Require().True(calls[0].hasDeadline, "Receive context should carry a deadline") + s.WithinDuration(calls[0].entered.Add(tt.expect), calls[0].deadline, 200*time.Millisecond) + }) + } +} + +func (s *QueueTestSuite) TestReceiveProcessesJobWithConfiguredTimeout() { + s.timeoutDriver = newReceiveTimeoutDriver() + s.registerTimeoutConnection(true, "1s") + s.timeoutDriver.deliver = true + + s.runTimeoutWorker() // stop waits for Run, which waits for Handle/Delete to complete + + s.Equal([]any{"timeout-job"}, jobs.TestResult) + job := s.timeoutDriver.delivered() + s.Require().NotNil(job) + s.Equal(1, job.deletes()) +} + +func (s *QueueTestSuite) TestBlockingReceiveIsBoundedByConfiguredTimeout() { + s.timeoutDriver = newReceiveTimeoutDriver() + s.registerTimeoutConnection(true, "300ms") + s.timeoutDriver.block = true + s.timeoutDriver.deliver = true + + stop := s.startTimeoutWorker() + defer stop() + + select { + case <-s.timeoutDriver.firstReceive: + case <-time.After(2 * time.Second): + s.FailNow("Receive was not called") + } + + // The first Receive blocks until the 300ms deadline, so a second Receive + // only happens if the configured timeout actually bounded the first one. + select { + case <-s.timeoutDriver.secondReceive: + case <-time.After(2 * time.Second): + s.FailNow("worker did not issue a second Receive") + } + + // Stop explicitly so the post-timeout delivery has been processed and + // acknowledged before asserting; the deferred stop is then a no-op. + stop() + + s.Equal([]any{"timeout-job"}, jobs.TestResult) + job := s.timeoutDriver.delivered() + s.Require().NotNil(job) + s.Equal(1, job.deletes()) + + // Asserting both calls guards against a regression that builds the timeout + // context once and reuses it across loop iterations. + calls := s.timeoutDriver.calls() + s.Require().GreaterOrEqual(len(calls), 2) + s.Equal("critical", calls[0].queue) + s.Equal(3, calls[0].count) + s.Require().True(calls[0].hasDeadline, "Receive context should carry a deadline") + s.WithinDuration(calls[0].entered.Add(300*time.Millisecond), calls[0].deadline, 100*time.Millisecond) + s.Equal("critical", calls[1].queue) + s.Equal(3, calls[1].count) + s.Require().True(calls[1].hasDeadline, "Receive context should carry a deadline") + s.WithinDuration(calls[1].entered.Add(300*time.Millisecond), calls[1].deadline, 100*time.Millisecond) +} + +func (s *QueueTestSuite) TestShutdownCancelsBlockedReceive() { + s.timeoutDriver = newReceiveTimeoutDriver() + s.registerTimeoutConnection(true, "30s") + s.timeoutDriver.block = true + + stop := s.startTimeoutWorker() + defer stop() + + select { + case <-s.timeoutDriver.firstReceive: + case <-time.After(2 * time.Second): + s.FailNow("Receive was not called") + } + + start := time.Now() + stop() + s.Less(time.Since(start), time.Second, "Shutdown should cancel the in-flight Receive") +} + +func (s *QueueTestSuite) TestReceiveCapableDriverUsesPopForQueueList() { + s.timeoutDriver = newReceiveTimeoutDriver() + s.registerTimeoutConnection(false, nil) + + // A comma-separated queue list is consumed via Pop to preserve priority + // order, so the receive path must not be taken even though the driver + // implements DriverWithReceive. + worker := facades.Queue().Worker(contractsqueue.Args{ + Connection: timeoutTestConnection, + Queue: "default,high", + Concurrent: 1, + }) + stop := s.startWorker(worker) + defer stop() + + select { + case <-s.timeoutDriver.popSignal: + case <-time.After(2 * time.Second): + s.FailNow("Pop was not called") + } + + s.Require().Eventually(func() bool { + return len(s.timeoutDriver.pops()) >= 2 + }, 2*time.Second, 10*time.Millisecond, "expected both queues to be polled") + + s.Equal([]string{"default", "high"}, s.timeoutDriver.pops()[:2]) + s.Empty(s.timeoutDriver.calls(), "Receive must not be called for a multi-queue worker") +} + +// registerTimeoutConnection installs the test-only receive connection for the +// current case. The worker reads `timeout` and `via` once at construction, so +// the connection must be registered before facades.Queue().Worker(...). +func (s *QueueTestSuite) registerTimeoutConnection(includeTimeout bool, timeout any) { + driver := s.timeoutDriver + connection := map[string]any{ + "driver": "custom", + "connection": "default", + "queue": "default", + "concurrent": 1, + "via": func() (contractsqueue.Driver, error) { + return driver, nil + }, + } + if includeTimeout { + connection["timeout"] = timeout + } + facades.Config().Add("queue.connections."+timeoutTestConnection, connection) +} + +// startTimeoutWorker starts a worker on the test connection. The queue and +// concurrency differ from the connection defaults so the receive assertions +// prove the worker args are forwarded rather than read from config. +func (s *QueueTestSuite) startTimeoutWorker() (stop func()) { + worker := facades.Queue().Worker(contractsqueue.Args{ + Connection: timeoutTestConnection, + Queue: "critical", + Concurrent: 3, + }) + + return s.startWorker(worker) +} + +// runTimeoutWorker starts a worker on the test connection, waits for the first +// Receive, then stops the worker. +func (s *QueueTestSuite) runTimeoutWorker() { + stop := s.startTimeoutWorker() + defer stop() + + select { + case <-s.timeoutDriver.firstReceive: + case <-time.After(2 * time.Second): + s.FailNow("Receive was not called") + } +} + var ( testQueueArgs = []contractsqueue.Arg{ {