mirror of
https://github.com/acaloiaro/neoq
synced 2026-07-21 10:12:18 +00:00
Initial commit
This commit is contained in:
Adriano Caloiaro
commit
ad6f6c0387
12 changed files with 1474 additions and 0 deletions
15
.gitignore
vendored
Normal file
15
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
# Binaries for programs and plugins
|
||||
*.exe
|
||||
*.exe~
|
||||
*.dll
|
||||
*.so
|
||||
*.dylib
|
||||
|
||||
# Test binary, built with `go test -c`
|
||||
*.test
|
||||
|
||||
# Output of the go coverage tool, specifically when used with LiteIDE
|
||||
*.out
|
||||
|
||||
# Dependency directories (remove the comment below to include it)
|
||||
# vendor/
|
||||
5
LICENSE.txt
Normal file
5
LICENSE.txt
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
Copyright (c) Zenity Labs LLC
|
||||
|
||||
Neoq is an Open Source project licensed under the terms of
|
||||
the ALGPLv3 license. Please see <https://www.gnu.org/licenses/agpl-3.0.en.html>
|
||||
for license text.
|
||||
68
README.md
Normal file
68
README.md
Normal file
|
|
@ -0,0 +1,68 @@
|
|||
# Neoq
|
||||
|
||||
Background job processing built on Postgres for Go
|
||||
|
||||
# Installation
|
||||
|
||||
`go get github.com/acaloiaro/neoq`
|
||||
|
||||
# About
|
||||
|
||||
Neoq is a background job framework for Go applications. Its purpose is to minimize the infrastructure necessary to run production applications. It does so by implementing queue durability with modular backends, rather than introducing a strict dependency on a particular backend such as Redis.
|
||||
|
||||
The initial backend is based on Postgres since many applications already require a ralational database, and Postgres is an excellent one.
|
||||
|
||||
Neoq does not aim to be the _fastest_ background job processor. It aims to be _fast_, _reliable_, and demand a minimal infrastructure footprint.
|
||||
|
||||
# Features
|
||||
|
||||
- **Postgres-backed** job processing
|
||||
- **Job uniqueness**: jobs are fingerprinted based on their payload and status to prevent job duplication (multiple unprocessed jobs with the same payload cannot be queued)
|
||||
- **Retries**: Jobs may be retried a configurable number of times with exponential backoff and jitter to prevent thundering herds
|
||||
- **Deadlines**: Queue handlers can be configured with per-job time deadlines with millisecond accuracy
|
||||
- **Configurable transaction idle time**: Don't let your background worker transactions run away with db resources. By default, worker transactions may idle for 60 seconds.
|
||||
- **Future Jobs**: Jobs can be scheduled either for the future or immediate execution
|
||||
- **Concurrency**: Concurrency is configurable for every queue
|
||||
|
||||
# Getting Started
|
||||
|
||||
Getting started is as simple as declaring queue handlers and adding jobs.
|
||||
|
||||
Additional documentation can be found in the wiki: https://github.com/acaloiaro/neoq/wiki
|
||||
|
||||
Error handling in this section is excluded for simplicity.
|
||||
|
||||
## 1. Add queue handlers
|
||||
|
||||
Queue handlers listen for Jobs on queues. Jobs may consist of any payload that is JSON-serializable. Payloads are stored in Postgres as `jsonb` fields.
|
||||
|
||||
Queue Handlers are simple Go functions that accept a `Context` parameter.
|
||||
|
||||
**Example**: Add a listener on the `hello_world` queue
|
||||
|
||||
```go
|
||||
nq, _ := neoq.New("postgres://username:password@127.0.0.1:5432/neoq?sslmode=disable")
|
||||
nq.Listen("hello_world", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
j, err := neoq.JobFromContext(ctx)
|
||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||
return
|
||||
}))
|
||||
```
|
||||
|
||||
## 2. Queue jobs
|
||||
|
||||
**Example**: Add a "Hello World" job to the `hello_world` queue
|
||||
|
||||
```go
|
||||
nq, _ := neoq.New("postgres://username:password@127.0.0.1:5432/neoq?sslmode=disable")
|
||||
jid, _ := nq.Enqueue(neoq.Job{
|
||||
Queue: "hello_world",
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello world",
|
||||
},
|
||||
})
|
||||
|
||||
```
|
||||
|
||||
|
||||
|
||||
3
examples/README.md
Normal file
3
examples/README.md
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
# Examples
|
||||
|
||||
This collection of examples is intended as a starting point for adding Neoq to applications. It is not intended to be comprehensive, but as a complement to documentation.
|
||||
27
examples/add_future_job/main.go
Normal file
27
examples/add_future_job/main.go
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
)
|
||||
|
||||
func main() {
|
||||
const queue = "foobar"
|
||||
nq, _ := neoq.New("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable", neoq.TransactionTimeoutOpt(1000))
|
||||
|
||||
// Add a job that will execute 1 hour from now
|
||||
jobID, err := nq.Enqueue(neoq.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello, future world",
|
||||
},
|
||||
RunAfter: time.Now().Add(1 * time.Hour),
|
||||
})
|
||||
if err != nil {
|
||||
log.Println("error adding job", err)
|
||||
}
|
||||
|
||||
log.Println("added job:", jobID)
|
||||
}
|
||||
54
examples/add_job_with_custom_concurrnecy/main.go
Normal file
54
examples/add_job_with_custom_concurrnecy/main.go
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
)
|
||||
|
||||
func main() {
|
||||
var err error
|
||||
const queue = "foobar"
|
||||
//
|
||||
nq, _ := neoq.New("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable")
|
||||
|
||||
// we use a done channel here to make sure that our test doesn't exit before the job finishes running
|
||||
// this is probably not a pattern you want to use in production jobs and you see it here only for testing reasons
|
||||
done := make(chan bool)
|
||||
|
||||
// Concurrency and other options may be set on handlers both during creation (Option 1), or after the fact (Option 2)
|
||||
// Option 1: add options when creating the handler
|
||||
handler := neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
var j *neoq.Job
|
||||
j, err = neoq.JobFromContext(ctx)
|
||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||
done <- true
|
||||
return
|
||||
}, neoq.WithConcurrency(8))
|
||||
|
||||
// Option 2: Set options after the handler is created
|
||||
handler = handler.WithOption(neoq.WithConcurrency(8))
|
||||
|
||||
err = nq.Listen(queue, handler)
|
||||
if err != nil {
|
||||
log.Println("error listening to queue", err)
|
||||
}
|
||||
|
||||
// Add a job that will execute 1 hour from now
|
||||
_, err = nq.Enqueue(neoq.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello, world",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
log.Println("error adding job", err)
|
||||
}
|
||||
|
||||
<-done
|
||||
|
||||
// job's status will be 'failed' and 'error' will be 'job exceeded its 10ms deadline'
|
||||
// until either the job's Sleep statement is decreased/removed or the handler's deadline is increased
|
||||
// this job will continue to to fail and ultimately land on the dead jobs queue
|
||||
}
|
||||
57
examples/add_job_with_deadline/main.go
Normal file
57
examples/add_job_with_deadline/main.go
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
)
|
||||
|
||||
func main() {
|
||||
var err error
|
||||
const queue = "foobar"
|
||||
//
|
||||
nq, _ := neoq.New(
|
||||
"postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable",
|
||||
neoq.TransactionTimeoutOpt(1000), // transactions may be idle up to one second
|
||||
)
|
||||
|
||||
// we use a done channel here to make sure that our test doesn't exit before the job finishes running
|
||||
// this is probably not a pattern you want to use in production jobs and you see it here only for testing reasons
|
||||
done := make(chan bool)
|
||||
|
||||
handler := neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
var j *neoq.Job
|
||||
time.Sleep(1 * time.Second)
|
||||
j, err = neoq.JobFromContext(ctx)
|
||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||
done <- true
|
||||
return
|
||||
})
|
||||
|
||||
// this 10ms deadline will cause our job that sleeps for 1s to fail
|
||||
handler = handler.WithOption(neoq.WithDeadline(10 * time.Millisecond))
|
||||
|
||||
err = nq.Listen(queue, handler)
|
||||
if err != nil {
|
||||
log.Println("error listening to queue", err)
|
||||
}
|
||||
|
||||
// Add a job that will execute 1 hour from now
|
||||
_, err = nq.Enqueue(neoq.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello, world",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
log.Println("error adding job", err)
|
||||
}
|
||||
|
||||
<-done
|
||||
|
||||
// job's status will be 'failed' and 'error' will be 'job exceeded its 10ms deadline'
|
||||
// until either the job's Sleep statement is decreased/removed or the handler's deadline is increased
|
||||
// this job will continue to to fail and ultimately land on the dead jobs queue
|
||||
}
|
||||
35
examples/listen_for_jobs/main.go
Normal file
35
examples/listen_for_jobs/main.go
Normal file
|
|
@ -0,0 +1,35 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
)
|
||||
|
||||
func main() {
|
||||
var err error
|
||||
const queue = "foobar"
|
||||
//
|
||||
nq, _ := neoq.New(
|
||||
"postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable",
|
||||
neoq.TransactionTimeoutOpt(1000), // transactions may be idle up to one second
|
||||
)
|
||||
|
||||
handler := neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
var j *neoq.Job
|
||||
time.Sleep(1 * time.Second)
|
||||
j, err = neoq.JobFromContext(ctx)
|
||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||
return
|
||||
})
|
||||
|
||||
err = nq.Listen(queue, handler)
|
||||
if err != nil {
|
||||
log.Println("error listening to queue", err)
|
||||
}
|
||||
|
||||
// this code will exit quickly since since Listen() is asynchronous
|
||||
// real applications should call Listen() on startup for every queue that needs to be handled
|
||||
}
|
||||
22
go.mod
Normal file
22
go.mod
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
module github.com/acaloiaro/neoq
|
||||
|
||||
go 1.19
|
||||
|
||||
require (
|
||||
github.com/guregu/null v4.0.0+incompatible
|
||||
github.com/jackc/pgconn v1.13.0
|
||||
github.com/jackc/pgx/v5 v5.2.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/jackc/chunkreader/v2 v2.0.1 // indirect
|
||||
github.com/jackc/pgio v1.0.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgproto3/v2 v2.3.1 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
|
||||
github.com/jackc/puddle/v2 v2.1.2 // indirect
|
||||
go.uber.org/atomic v1.10.0 // indirect
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
|
||||
golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 // indirect
|
||||
golang.org/x/text v0.3.8 // indirect
|
||||
)
|
||||
148
go.sum
Normal file
148
go.sum
Normal file
|
|
@ -0,0 +1,148 @@
|
|||
github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
|
||||
github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
|
||||
github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
|
||||
github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
|
||||
github.com/guregu/null v4.0.0+incompatible h1:4zw0ckM7ECd6FNNddc3Fu4aty9nTlpkkzH7dPn4/4Gw=
|
||||
github.com/guregu/null v4.0.0+incompatible/go.mod h1:ePGpQaN9cw0tj45IR5E5ehMvsFlLlQZAkkOXZurJ3NM=
|
||||
github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9QQ7T3kePo=
|
||||
github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
|
||||
github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
|
||||
github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
|
||||
github.com/jackc/pgconn v0.0.0-20190420214824-7e0022ef6ba3/go.mod h1:jkELnwuX+w9qN5YIfX0fl88Ehu4XC3keFuOJJk9pcnA=
|
||||
github.com/jackc/pgconn v0.0.0-20190824142844-760dd75542eb/go.mod h1:lLjNuW/+OfW9/pnVKPazfWOgNfH2aPem8YQ7ilXGvJE=
|
||||
github.com/jackc/pgconn v0.0.0-20190831204454-2fabfa3c18b7/go.mod h1:ZJKsE/KZfsUgOEh9hBm+xYTstcNHg7UPMVJqRfQxq4s=
|
||||
github.com/jackc/pgconn v1.8.0/go.mod h1:1C2Pb36bGIP9QHGBYCjnyhqu7Rv3sGshaQUvmfGIB/o=
|
||||
github.com/jackc/pgconn v1.9.0/go.mod h1:YctiPyvzfU11JFxoXokUOOKQXQmDMoJL9vJzHH8/2JY=
|
||||
github.com/jackc/pgconn v1.13.0 h1:3L1XMNV2Zvca/8BYhzcRFS70Lr0WlDg16Di6SFGAbys=
|
||||
github.com/jackc/pgconn v1.13.0/go.mod h1:AnowpAqO4CMIIJNZl2VJp+KrkAZciAkhEl0W0JIobpI=
|
||||
github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
|
||||
github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8=
|
||||
github.com/jackc/pgmock v0.0.0-20190831213851-13a1b77aafa2/go.mod h1:fGZlG77KXmcq05nJLRkk0+p82V8B8Dw8KN2/V9c/OAE=
|
||||
github.com/jackc/pgmock v0.0.0-20201204152224-4fe30f7445fd/go.mod h1:hrBW0Enj2AZTNpt/7Y5rr2xe/9Mn757Wtb2xeBzPv2c=
|
||||
github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5Wi/+Zz7xoE5ALHsRQlOctkOiHc=
|
||||
github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgproto3 v1.1.0/go.mod h1:eR5FA3leWg7p9aeAqi37XOTgTIbkABlvcPB3E5rlc78=
|
||||
github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190420180111-c116219b62db/go.mod h1:bhq50y+xrl9n5mRYyCBFKkpRVTLYJVWeCc+mEAI3yXA=
|
||||
github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190609003834-432c2951c711/go.mod h1:uH0AWtUmuShn0bcesswc4aBTWGvw0cAxIJp+6OB//Wg=
|
||||
github.com/jackc/pgproto3/v2 v2.0.0-rc3/go.mod h1:ryONWYqW6dqSg1Lw6vXNMXoBJhpzvWKnT95C46ckYeM=
|
||||
github.com/jackc/pgproto3/v2 v2.0.0-rc3.0.20190831210041-4c03ce451f29/go.mod h1:ryONWYqW6dqSg1Lw6vXNMXoBJhpzvWKnT95C46ckYeM=
|
||||
github.com/jackc/pgproto3/v2 v2.0.6/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
|
||||
github.com/jackc/pgproto3/v2 v2.1.1/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
|
||||
github.com/jackc/pgproto3/v2 v2.3.1 h1:nwj7qwf0S+Q7ISFfBndqeLwSwxs+4DPsbRFjECT1Y4Y=
|
||||
github.com/jackc/pgproto3/v2 v2.3.1/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b h1:C8S2+VttkHFdOOCXJe+YGfa4vHYwlt4Zx+IVXQ97jYg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b/go.mod h1:vsD4gTJCa9TptPL8sPkXrLZ+hDuNrZCnj29CQpr4X1E=
|
||||
github.com/jackc/pgtype v0.0.0-20190421001408-4ed0de4755e0/go.mod h1:hdSHsc1V01CGwFsrv11mJRHWJ6aifDLfdV3aVjFF0zg=
|
||||
github.com/jackc/pgtype v0.0.0-20190824184912-ab885b375b90/go.mod h1:KcahbBH1nCMSo2DXpzsoWOAfFkdEtEJpPbVLq8eE+mc=
|
||||
github.com/jackc/pgtype v0.0.0-20190828014616-a8802b16cc59/go.mod h1:MWlu30kVJrUS8lot6TQqcg7mtthZ9T0EoIBFiJcmcyw=
|
||||
github.com/jackc/pgx/v4 v4.0.0-20190420224344-cc3461e65d96/go.mod h1:mdxmSJJuR08CZQyj1PVQBHy9XOp5p8/SHH6a0psbY9Y=
|
||||
github.com/jackc/pgx/v4 v4.0.0-20190421002000-1b8f0016e912/go.mod h1:no/Y67Jkk/9WuGR0JG/JseM9irFbnEPbuWV2EELPNuM=
|
||||
github.com/jackc/pgx/v4 v4.0.0-pre1.0.20190824185557-6972a5742186/go.mod h1:X+GQnOEnf1dqHGpw7JmHqHc1NxDoalibchSk9/RWuDc=
|
||||
github.com/jackc/pgx/v5 v5.2.0 h1:NdPpngX0Y6z6XDFKqmFQaE+bCtkqzvQIOt1wvBlAqs8=
|
||||
github.com/jackc/pgx/v5 v5.2.0/go.mod h1:Ptn7zmohNsWEsdxRawMzk3gaKma2obW+NWTnKa0S4nk=
|
||||
github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
|
||||
github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
|
||||
github.com/jackc/puddle/v2 v2.1.2 h1:0f7vaaXINONKTsxYDn4otOAiJanX/BMeAtY//BXqzlg=
|
||||
github.com/jackc/puddle/v2 v2.1.2/go.mod h1:2lpufsF5mRHO6SuZkm0fNYxM6SWHfvyFj62KwNzgels=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/pty v1.1.8/go.mod h1:O1sed60cT9XZ5uDucP5qwvh+TE3NnUj51EiZO/lmSfw=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
|
||||
github.com/lib/pq v1.1.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
|
||||
github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
|
||||
github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
|
||||
github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
|
||||
github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
|
||||
github.com/rs/zerolog v1.13.0/go.mod h1:YbFCdg8HfsridGWAh22vktObvhZbQsZXe4/zB0OKkWU=
|
||||
github.com/rs/zerolog v1.15.0/go.mod h1:xYTKnLHcpfU2225ny5qZjxnj9NvkumZYjJHlAThCjNc=
|
||||
github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
|
||||
github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
|
||||
github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
|
||||
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
|
||||
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
|
||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
|
||||
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
||||
github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q=
|
||||
go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
|
||||
go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
|
||||
go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
|
||||
go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
|
||||
go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
|
||||
go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
|
||||
go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190411191339-88737f569e3a/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
|
||||
golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
|
||||
golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
|
||||
golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
|
||||
golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ffipvEDX2pl7E1TdqLqIc=
|
||||
golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||
golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY=
|
||||
golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20190425163242-31fd60d6bfdc/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
|
||||
golang.org/x/tools v0.0.0-20190823170909-c4a336ef6a2f/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/xerrors v0.0.0-20190410155217-1f06c39b4373/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/inconshreveable/log15.v2 v2.0.0-20180818164646-67afb5ed74ec/go.mod h1:aPpfJ7XW+gOuirDoZ8gHhLh3kZ1B08FtV2bbmy7Jv3s=
|
||||
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
958
neoq.go
Normal file
958
neoq.go
Normal file
|
|
@ -0,0 +1,958 @@
|
|||
package neoq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"math"
|
||||
"os"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"math/rand"
|
||||
|
||||
"github.com/guregu/null"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type contextKey int
|
||||
|
||||
var varsKey contextKey
|
||||
|
||||
const (
|
||||
JobStatusNew = "new"
|
||||
JobStatusProcessed = "processed"
|
||||
JobStatusFailed = "failed"
|
||||
DefaultTransactionTimeout int = 60000 //ms
|
||||
DefaultHandlerDeadline = 30000 //ms
|
||||
DuplicateJobID = -1
|
||||
|
||||
PendingJobIDQuery = `SELECT id
|
||||
FROM neoq_jobs
|
||||
WHERE queue = $1
|
||||
AND status NOT IN ('processed')
|
||||
AND run_after <= NOW()
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1`
|
||||
PendingJobQuery = `SELECT id,fingerprint,queue,status,payload,retries,max_retries,run_after,ran_at,created_at,error
|
||||
FROM neoq_jobs
|
||||
WHERE id = $1
|
||||
AND status NOT IN ('processed')
|
||||
AND run_after <= NOW()
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1`
|
||||
FutureJobQuery = `SELECT id,run_after
|
||||
FROM neoq_jobs
|
||||
WHERE queue = $1
|
||||
AND status NOT IN ('processed')
|
||||
AND run_after > NOW()
|
||||
ORDER BY run_after ASC
|
||||
LIMIT 100
|
||||
FOR UPDATE SKIP LOCKED`
|
||||
)
|
||||
|
||||
// Config configures Neoq's general behavior
|
||||
type Config struct {
|
||||
// the time that a worker's transaction may be idle before its underlying connection is closed
|
||||
idleTxTimeout int
|
||||
}
|
||||
|
||||
// ConfigOption is function that sets optional Neoq configuration
|
||||
type ConfigOption func(n Neoq)
|
||||
|
||||
// Job contains all the data pertaining to a job
|
||||
type Job struct {
|
||||
ID int64 `db:"id"`
|
||||
Fingerprint string `db:"fingerprint"` // A sha256 sum of the job's payload
|
||||
Status string `db:"status"` // The status of the job
|
||||
Queue string `db:"queue"` // The queue the job is on
|
||||
Payload map[string]any `db:"payload"` // JSON job payload for more complex jobs
|
||||
RunAfter time.Time `db:"run_after"` // The time after which the job is elligible to be picked up by a worker
|
||||
RanAt null.Time `db:"ran_at"` // The last time the job ran
|
||||
Error null.String `db:"error"` // The last error the job elicited
|
||||
Retries int `db:"retries"` // The number of times the job has retried
|
||||
MaxRetries int `db:"max_retries"` // The maximum number of times the job can retry
|
||||
CreatedAt time.Time `db:"created_at"` // The time the job was created
|
||||
}
|
||||
|
||||
// handlerCtxVars are variables passed to every Handler context
|
||||
type handlerCtxVars struct {
|
||||
job *Job
|
||||
tx pgx.Tx
|
||||
}
|
||||
|
||||
// TransactionTimeoutOpt sets the time that a worker's transaction may be idle before its underlying connection is
|
||||
// closed
|
||||
// The timeout is the number of milliseconds that a transaction may sit idle before postgres terminates the
|
||||
// transaction's underlying connection. The timeout should be longer than your longest job takes to complete. If set
|
||||
// too short, job state will become unpredictable, e.g. retry counts may become incorrect.
|
||||
func TransactionTimeoutOpt(txTimeout int) ConfigOption {
|
||||
return func(n Neoq) {
|
||||
n.Config().idleTxTimeout = txTimeout
|
||||
|
||||
// implementation is a pgWorker; sets the AfterConnect on its pool
|
||||
if pgw, ok := n.(*pgWorker); ok {
|
||||
// if the worker already has a pool configured, then setting the transaction timeout prints a warning and is a
|
||||
// no-op
|
||||
if pgw.pool != nil {
|
||||
log.Println("create a new Neoq instance to set transaction timeout")
|
||||
return
|
||||
}
|
||||
|
||||
var err error
|
||||
var poolConfig *pgxpool.Config
|
||||
|
||||
poolConfig, err = pgxpool.ParseConfig(pgw.dbConnectString)
|
||||
if err != nil || pgw.dbConnectString == "" {
|
||||
return
|
||||
}
|
||||
|
||||
// ensure that workers don't consume connections with idle transactions
|
||||
poolConfig.AfterConnect = func(ctx context.Context, conn *pgx.Conn) (err error) {
|
||||
var query string
|
||||
if n.Config().idleTxTimeout > 0 {
|
||||
query = fmt.Sprintf("SET idle_in_transaction_session_timeout = '%dms'", n.Config().idleTxTimeout)
|
||||
} else {
|
||||
// there is no limit to the amount of time a worker's transactions may be idle
|
||||
query = "SET idle_in_transaction_session_timeout = 0"
|
||||
}
|
||||
_, err = conn.Exec(ctx, query)
|
||||
return
|
||||
}
|
||||
|
||||
pgw.pool, err = pgxpool.NewWithConfig(context.Background(), poolConfig)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
// JobFromContext fetches the job from a context if it is set
|
||||
func JobFromContext(ctx context.Context) (j *Job, err error) {
|
||||
if v, ok := ctx.Value(varsKey).(handlerCtxVars); ok {
|
||||
j = v.job
|
||||
} else {
|
||||
err = errors.New("context does not have a Job set")
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// withHandlerContext creates a new context with the job and transaction set
|
||||
func withHandlerContext(ctx context.Context, v handlerCtxVars) context.Context {
|
||||
return context.WithValue(ctx, varsKey, v)
|
||||
}
|
||||
|
||||
// txFromContext gets the transaction from a context, if the the transaction is already set
|
||||
func txFromContext(ctx context.Context) (t pgx.Tx, err error) {
|
||||
if v, ok := ctx.Value(varsKey).(handlerCtxVars); ok {
|
||||
t = v.tx
|
||||
} else {
|
||||
err = errors.New("context does not have a Tx set")
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// Neoq interface is he primary API for Neoq
|
||||
type Neoq interface {
|
||||
// Enqueue queues jobs to be executed asynchronously
|
||||
Enqueue(job Job) (jobID int64, err error)
|
||||
|
||||
// Listen listens for jobs on a queue and processes them with the given handler
|
||||
Listen(queue string, h Handler) (err error)
|
||||
|
||||
// Shutdown halts the worker
|
||||
Shutdown() bool
|
||||
|
||||
// WithOption configures the instance with optional configuration
|
||||
WithConfigOpt(opt ConfigOption) Neoq
|
||||
|
||||
// Config retrieves an instance's configuration
|
||||
Config() *Config
|
||||
}
|
||||
|
||||
// HandlerFunc is a function that Handlers execute for every Job on a queue
|
||||
type HandlerFunc func(ctx context.Context) error
|
||||
|
||||
// Handler handles jobs on a queue
|
||||
type Handler struct {
|
||||
Deadline time.Duration
|
||||
Handle HandlerFunc
|
||||
MaxRetries int
|
||||
Concurrency int
|
||||
}
|
||||
|
||||
// HandlerOption is function that sets optional configuration for Handlers
|
||||
type HandlerOption func(w *Handler)
|
||||
|
||||
// WithOption returns the handler with the given options set
|
||||
func (h Handler) WithOption(opt HandlerOption) (handler Handler) {
|
||||
opt(&h)
|
||||
return h
|
||||
}
|
||||
|
||||
// WithDeadline configures handlers with a deadline for every job that it excutes
|
||||
// The deadline is the amount of time (ms) that can be spent executing the handler's HandlerFunction
|
||||
// when the deadline is exceeded, jobs are failed and begin the retry phase of their lifecycle
|
||||
func WithDeadline(d time.Duration) HandlerOption {
|
||||
return func(h *Handler) {
|
||||
h.Deadline = d
|
||||
}
|
||||
}
|
||||
|
||||
// WithConcurrency configures Neoq handlers to process jobs concurrently
|
||||
// the default concurrency is the number of (v)CPUs on the machine running Neoq
|
||||
func WithConcurrency(c int) HandlerOption {
|
||||
return func(h *Handler) {
|
||||
h.Concurrency = c
|
||||
}
|
||||
}
|
||||
|
||||
// NewHandler creates a new queue handler
|
||||
func NewHandler(f HandlerFunc, opts ...HandlerOption) (h Handler) {
|
||||
h = Handler{Handle: f, Concurrency: runtime.NumCPU() - 1}
|
||||
|
||||
for _, opt := range opts {
|
||||
opt(&h)
|
||||
}
|
||||
|
||||
// always set a job deadline if none is set
|
||||
if h.Deadline == 0 {
|
||||
h.Deadline = time.Duration(DefaultHandlerDeadline * time.Millisecond)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// pgWorker is a concrete Neoq implementation based on Postgres, using pgx/pgxpool
|
||||
type pgWorker struct {
|
||||
config *Config
|
||||
dbConnectString string
|
||||
idleTxTimeout int
|
||||
listenConn *pgx.Conn
|
||||
pool *pgxpool.Pool
|
||||
handlers map[string]Handler // a map of queue names to queue handlers
|
||||
mu *sync.Mutex // mutext to protect mutating state on a pgWorker
|
||||
futureJobs map[int64]time.Time // map of future job IDs to their due time
|
||||
}
|
||||
|
||||
// New creates a new Neoq instance for listening to queues and enqueing new jobs
|
||||
//
|
||||
// Connection strings may be a URL or DSN-style connection string to neoq's database. The connection string supports multiple
|
||||
// options that can be left empty to use defaults:
|
||||
//
|
||||
// options:
|
||||
// pool_max_conns: integer greater than 0
|
||||
// pool_min_conns: integer 0 or greater
|
||||
// pool_max_conn_lifetime: duration string
|
||||
// pool_max_conn_idle_time: duration string
|
||||
// pool_health_check_period: duration string
|
||||
// pool_max_conn_lifetime_jitter: duration string
|
||||
//
|
||||
// # Example DSN
|
||||
// user=worker password=secret host=workerdb.example.com port=5432 dbname=mydb sslmode=verify-ca pool_max_conns=10
|
||||
//
|
||||
// # Example URL
|
||||
// postgres://worker:secret@workerdb.example.com:5432/mydb?sslmode=verify-ca&pool_max_conns=10
|
||||
//
|
||||
// Available ConfigOption
|
||||
// - WithTransactionTimeout(timeout int): configure the idle_in_transaction_timeout for the worker's database
|
||||
// connection(s)
|
||||
func New(connection string, opts ...ConfigOption) (n Neoq, err error) {
|
||||
|
||||
w := pgWorker{
|
||||
mu: &sync.Mutex{},
|
||||
config: &Config{},
|
||||
dbConnectString: connection,
|
||||
handlers: make(map[string]Handler),
|
||||
futureJobs: make(map[int64]time.Time),
|
||||
}
|
||||
|
||||
// Set all options
|
||||
for _, opt := range opts {
|
||||
opt(w)
|
||||
}
|
||||
|
||||
err = w.initializeDB()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if w.pool == nil {
|
||||
var poolConfig *pgxpool.Config
|
||||
poolConfig, err = pgxpool.ParseConfig(w.dbConnectString)
|
||||
if err != nil || w.dbConnectString == "" {
|
||||
return nil, errors.New("invalid connecton string: see documentation for valid connection strings")
|
||||
}
|
||||
|
||||
// ensure that workers don't consume connections with idle transactions
|
||||
poolConfig.AfterConnect = func(ctx context.Context, conn *pgx.Conn) (err error) {
|
||||
var query string
|
||||
if w.idleTxTimeout > 0 {
|
||||
query = fmt.Sprintf("SET idle_in_transaction_session_timeout = '%dms'", w.idleTxTimeout)
|
||||
} else {
|
||||
// there is no limit to the amount of time a worker's transactions may be idle
|
||||
query = "SET idle_in_transaction_session_timeout = 0"
|
||||
}
|
||||
_, err = conn.Exec(ctx, query)
|
||||
return
|
||||
}
|
||||
|
||||
w.pool, err = pgxpool.NewWithConfig(context.Background(), poolConfig)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
n = w
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (w pgWorker) Config() *Config {
|
||||
return w.config
|
||||
}
|
||||
|
||||
// initializeTables initializes the tables, types, and indicies necessary to operate Neoq
|
||||
func (w pgWorker) initializeDB() (err error) {
|
||||
var pgxCfg *pgx.ConnConfig
|
||||
var tx pgx.Tx
|
||||
ctx := context.Background()
|
||||
pgxCfg, err = pgx.ParseConfig(w.dbConnectString)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
dbName := pgxCfg.Database
|
||||
pgxCfg.Database = ""
|
||||
conn, err := pgx.ConnectConfig(context.Background(), pgxCfg)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Unable to connect to database: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer conn.Close(ctx)
|
||||
|
||||
var dbExists bool
|
||||
dbExistsQ := fmt.Sprintf(`SELECT EXISTS (SELECT datname FROM pg_catalog.pg_database WHERE datname = '%s');`, dbName)
|
||||
rows, err := conn.Query(context.Background(), dbExistsQ)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to determne if jobs table exists: %v", err)
|
||||
return
|
||||
}
|
||||
for rows.Next() {
|
||||
err = rows.Scan(&dbExists)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to determine if jobs table exists: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
conn.Close(ctx)
|
||||
pgxCfg.Database = dbName
|
||||
conn, err = pgx.ConnectConfig(context.Background(), pgxCfg)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Unable to connect to database: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer conn.Close(ctx)
|
||||
|
||||
if !dbExists {
|
||||
createDBQ := "CREATE DATABASE neoq"
|
||||
_, err := conn.Exec(ctx, createDBQ)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to create neoq database: %v", err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
tx, err = conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer tx.Rollback(ctx) // rollback has no effect if the transaction has been committed
|
||||
|
||||
jobsTableExistsQ := `SELECT EXISTS (SELECT FROM
|
||||
pg_tables
|
||||
WHERE
|
||||
schemaname = 'public' AND
|
||||
tablename = 'neoq_jobs'
|
||||
);`
|
||||
rows, err = tx.Query(context.Background(), jobsTableExistsQ)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to determne if jobs table exists: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
var tablesInitialized bool
|
||||
for rows.Next() {
|
||||
err = rows.Scan(&tablesInitialized)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to determine if jobs table exists: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
if !tablesInitialized {
|
||||
createTablesQ := `
|
||||
CREATE TYPE job_status AS ENUM (
|
||||
'new',
|
||||
'processed',
|
||||
'failed'
|
||||
);
|
||||
|
||||
CREATE TABLE neoq_dead_jobs (
|
||||
id SERIAL NOT NULL,
|
||||
fingerprint text NOT NULL,
|
||||
queue text NOT NULL,
|
||||
status job_status NOT NULL default 'failed',
|
||||
payload jsonb,
|
||||
retries integer,
|
||||
max_retries integer,
|
||||
created_at timestamp with time zone DEFAULT now(),
|
||||
error text
|
||||
);
|
||||
|
||||
CREATE TABLE neoq_jobs (
|
||||
id SERIAL NOT NULL,
|
||||
fingerprint text NOT NULL,
|
||||
queue text NOT NULL,
|
||||
status job_status NOT NULL default 'new',
|
||||
payload jsonb,
|
||||
retries integer default 0,
|
||||
max_retries integer default 23,
|
||||
run_after timestamp with time zone DEFAULT now(),
|
||||
ran_at timestamp with time zone,
|
||||
created_at timestamp with time zone DEFAULT now(),
|
||||
error text
|
||||
);
|
||||
|
||||
CREATE INDEX neoq_job_fetcher_idx ON neoq_jobs (id, status, run_after);
|
||||
CREATE INDEX neoq_jobs_fetcher_idx ON neoq_jobs (queue, status, run_after);
|
||||
`
|
||||
|
||||
_, err := tx.Exec(ctx, createTablesQ)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to create job status enum: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
err = tx.Commit(ctx)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// Enqueue queues jobs to be performed by workers
|
||||
func (w pgWorker) Enqueue(job Job) (jobID int64, err error) {
|
||||
ctx := context.Background()
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
log.Println("failed to acquire database connection to listen for new queue items")
|
||||
return
|
||||
}
|
||||
defer conn.Release()
|
||||
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Rollback is safe to call even if the tx is already closed, so if
|
||||
// the tx commits successfully, this is a no-op
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
// Make sure RunAfter is set to a non-zero value if not provided by the caller
|
||||
// if already set, schedule the future job
|
||||
now := time.Now()
|
||||
if job.RunAfter.IsZero() {
|
||||
job.RunAfter = now
|
||||
}
|
||||
|
||||
jobID, err = w.queueJob(ctx, tx, job)
|
||||
|
||||
// if the job ID is empty, then the job was skipped and no additional work is necessary
|
||||
if err != nil || jobID == DuplicateJobID {
|
||||
return
|
||||
}
|
||||
|
||||
// notify listeners that a new job has arrived if it's not a future job
|
||||
if job.RunAfter == now {
|
||||
_, err = tx.Exec(ctx, fmt.Sprintf("NOTIFY %s, '%d'", job.Queue, jobID))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
} else {
|
||||
w.mu.Lock()
|
||||
w.futureJobs[jobID] = job.RunAfter
|
||||
w.mu.Unlock()
|
||||
}
|
||||
|
||||
err = tx.Commit(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return jobID, nil
|
||||
}
|
||||
|
||||
// Listen listens for jobs on a queue and processes them
|
||||
func (w pgWorker) Listen(queue string, h Handler) (err error) {
|
||||
w.handlers[queue] = h
|
||||
|
||||
err = w.start(queue)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (w pgWorker) Shutdown() bool {
|
||||
w.pool.Close()
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (w pgWorker) WithConfigOpt(opt ConfigOption) Neoq {
|
||||
opt(&w)
|
||||
return &w
|
||||
}
|
||||
|
||||
// queueJob adds jobs to the queue, returning the job ID
|
||||
// Jobs that are not already fingerprinted are fingerprinted before being added
|
||||
// Duplicate jobs are not added to the queue. Any two unprocessed jobs with the same fingerprint are duplicates
|
||||
func (w pgWorker) queueJob(ctx context.Context, tx pgx.Tx, j Job) (jobID int64, err error) {
|
||||
|
||||
// fingerprint the job if it hasn't been fingerprinted already
|
||||
// a fingerprint is the md5 sum of: queue + payload
|
||||
if j.Fingerprint == "" {
|
||||
var js []byte
|
||||
js, err = json.Marshal(j.Payload)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
h := md5.New()
|
||||
io.WriteString(h, j.Queue)
|
||||
io.WriteString(h, string(js))
|
||||
j.Fingerprint = fmt.Sprintf("%x", h.Sum(nil))
|
||||
}
|
||||
|
||||
var rowCount int64
|
||||
countRow := tx.QueryRow(ctx, `SELECT COUNT(*) as row_count
|
||||
FROM neoq_jobs
|
||||
WHERE fingerprint = $1
|
||||
AND status NOT IN ('processed')`, j.Fingerprint)
|
||||
err = countRow.Scan(&rowCount)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// this is a duplicate job; skip it
|
||||
if rowCount > 0 {
|
||||
return DuplicateJobID, nil
|
||||
}
|
||||
|
||||
err = tx.QueryRow(ctx, `INSERT INTO neoq_jobs(queue, fingerprint, payload, run_after)
|
||||
VALUES ($1, $2, $3, $4) RETURNING id`,
|
||||
j.Queue, j.Fingerprint, j.Payload, j.RunAfter).Scan(&jobID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// moveToDeadQueue moves jobs from the pending queue to the dead queue
|
||||
func (w pgWorker) moveToDeadQueue(ctx context.Context, tx pgx.Tx, j *Job, jobErr error) (err error) {
|
||||
_, err = tx.Exec(ctx, "DELETE FROM neoq_jobs WHERE id = $1", j.ID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
_, err = tx.Exec(ctx, `INSERT INTO neoq_dead_jobs(id, queue, fingerprint, payload, retries, max_retries, error)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
|
||||
j.ID, j.Queue, j.Fingerprint, j.Payload, j.Retries, j.MaxRetries, jobErr.Error())
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// updateJob updates the status of jobs with: status, run time, error messages, and retries
|
||||
// if the retry count exceeds the maximum number of retries for the job, move the job to the dead jobs queue
|
||||
//
|
||||
// if `tx`'s underlying connection dies, it results in this function throwing an error and job status inaccurately
|
||||
// reflecting the status of the job and its number of retries. TODO: consider solutions to this problem, e.g. acquiring
|
||||
// a new connection in the event of connection failure
|
||||
func (w pgWorker) updateJob(ctx context.Context, jobErr error) (err error) {
|
||||
status := JobStatusProcessed
|
||||
errMsg := ""
|
||||
|
||||
if jobErr != nil {
|
||||
status = JobStatusFailed
|
||||
errMsg = jobErr.Error()
|
||||
}
|
||||
|
||||
var job *Job
|
||||
if job, err = JobFromContext(ctx); err != nil {
|
||||
return errors.New("unable to get job from context")
|
||||
}
|
||||
|
||||
var tx pgx.Tx
|
||||
if tx, err = txFromContext(ctx); err != nil {
|
||||
return errors.New("unable to get transaction from context")
|
||||
}
|
||||
|
||||
if job.Retries >= job.MaxRetries {
|
||||
err = w.moveToDeadQueue(ctx, tx, job, jobErr)
|
||||
return
|
||||
}
|
||||
|
||||
var runAfter time.Time
|
||||
if job.Retries > 0 && status == JobStatusFailed {
|
||||
runAfter = calculateBackoff(job.Retries)
|
||||
qstr := "UPDATE neoq_jobs SET ran_at = $1, error = $2, status = $3, retries = $4, run_after = $5 WHERE id = $6"
|
||||
_, err = tx.Exec(ctx, qstr, time.Now(), errMsg, status, job.Retries, runAfter, job.ID)
|
||||
} else if job.Retries > 0 && status != JobStatusFailed {
|
||||
qstr := "UPDATE neoq_jobs SET ran_at = $1, error = $2, status = $3, retries = $4 WHERE id = $5"
|
||||
_, err = tx.Exec(ctx, qstr, time.Now(), errMsg, status, job.Retries, job.ID)
|
||||
} else {
|
||||
qstr := "UPDATE neoq_jobs SET ran_at = $1, error = $2, status = $3 WHERE id = $4"
|
||||
_, err = tx.Exec(ctx, qstr, time.Now(), errMsg, status, job.ID)
|
||||
}
|
||||
|
||||
if err == nil && runAfter.Sub(time.Now()) > 0 {
|
||||
w.mu.Lock()
|
||||
w.futureJobs[job.ID] = runAfter
|
||||
w.mu.Unlock()
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// start starts a queue listener, processes pending job, and fires up goroutines to process future jobs
|
||||
func (w pgWorker) start(queue string) (err error) {
|
||||
|
||||
var handler Handler
|
||||
var ok bool
|
||||
if handler, ok = w.handlers[queue]; !ok {
|
||||
return fmt.Errorf("no handler for queue: %s", queue)
|
||||
}
|
||||
ctx := context.Background()
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// TODO: Give more thought to the implications of hijacking a conn from the pool here
|
||||
// should this connect not come from the pool, to avoid tainting it with connections that don't have an idle in
|
||||
// transaction time out set?
|
||||
w.listenConn = conn.Hijack()
|
||||
|
||||
listenJobChan := w.listen(ctx, queue) // listen for 'new' jobs
|
||||
pendingJobsChan := w.pendingJobs(ctx, queue) // process overdue jobs *at startup*
|
||||
|
||||
// process all future jobs and retries
|
||||
// TODO there are no illusions about the current future jobs system being robust
|
||||
// the current implementation is a brute force proof of concept that can certainly be improved upon
|
||||
go func() { w.scheduleFutureJobs(ctx, queue) }()
|
||||
|
||||
for i := 0; i < handler.Concurrency; i++ {
|
||||
go func() {
|
||||
var jobID int64
|
||||
|
||||
for {
|
||||
select {
|
||||
case jobID = <-listenJobChan:
|
||||
err = w.handleJob(ctx, jobID, handler)
|
||||
case jobID = <-pendingJobsChan:
|
||||
err = w.handleJob(ctx, jobID, handler)
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
if err.Error() == "no rows in result set" {
|
||||
err = nil
|
||||
} else {
|
||||
log.Println("error handling job", err)
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// removeFutureJob removes a future job from the in-memory list of jobs that will execute in the future
|
||||
func (w pgWorker) removeFutureJob(jobID int64) {
|
||||
if _, ok := w.futureJobs[jobID]; ok {
|
||||
w.mu.Lock()
|
||||
delete(w.futureJobs, jobID)
|
||||
w.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// initFutureJobs is intended to be run once to initialize the list of future jobs that must be monitored for
|
||||
// execution. it should be run only during system startup.
|
||||
func (w pgWorker) initFutureJobs(ctx context.Context, queue string) {
|
||||
rows, err := w.pool.Query(context.Background(), FutureJobQuery, queue)
|
||||
if err != nil {
|
||||
log.Println("error fetching future jobs list:", err)
|
||||
return
|
||||
}
|
||||
|
||||
var id int64
|
||||
var runAfter time.Time
|
||||
_, err = pgx.ForEachRow(rows, []any{&id, &runAfter}, func() error {
|
||||
w.mu.Lock()
|
||||
w.futureJobs[id] = runAfter
|
||||
w.mu.Unlock()
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// scheduleFutureJobs announces future jobs using NOTIFY on an interval
|
||||
func (w pgWorker) scheduleFutureJobs(ctx context.Context, queue string) {
|
||||
w.initFutureJobs(ctx, queue)
|
||||
|
||||
// check for new future jobs on an interval
|
||||
// TODO make this time configurable
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
|
||||
for {
|
||||
// loop over list of future jobs, scheduling goroutines to wait for jobs that are due within the next 30 seconds
|
||||
// TODO: Make 30 seconds configurable
|
||||
for jobID, runAfter := range w.futureJobs {
|
||||
at := runAfter.Sub(time.Now())
|
||||
if at <= time.Duration(30*time.Second) {
|
||||
w.removeFutureJob(jobID)
|
||||
go func(jid int64) {
|
||||
scheduleCh := time.After(at)
|
||||
<-scheduleCh
|
||||
w.announceJob(ctx, queue, jid)
|
||||
}(jobID)
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ticker.C:
|
||||
continue
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w pgWorker) announceJob(ctx context.Context, queue string, jobID int64) {
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
log.Println("failed to acquire database connection to schedule future job")
|
||||
return
|
||||
}
|
||||
defer conn.Release()
|
||||
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Rollback is safe to call even if the tx is already closed, so if
|
||||
// the tx commits successfully, this is a no-op
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
// notify listeners that a job is ready to run
|
||||
_, err = tx.Exec(ctx, fmt.Sprintf("NOTIFY %s, '%d'", queue, jobID))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = tx.Commit(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (w pgWorker) pendingJobs(ctx context.Context, queue string) (jobsCh chan int64) {
|
||||
jobsCh = make(chan int64)
|
||||
|
||||
// TODO Consider refactoring to use pgxpool.AcquireFunc()
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
log.Println("failed to acquire database connection to listen for new queue items:", err)
|
||||
return
|
||||
}
|
||||
|
||||
go func(ctx context.Context) {
|
||||
defer conn.Release()
|
||||
|
||||
for {
|
||||
jobID, err := w.getPendingJobID(ctx, conn, queue)
|
||||
if err != nil {
|
||||
if err.Error() != "no rows in result set" {
|
||||
log.Println(err)
|
||||
} else {
|
||||
// done fetching pending jobs
|
||||
break
|
||||
}
|
||||
} else {
|
||||
jobsCh <- jobID
|
||||
}
|
||||
}
|
||||
|
||||
}(ctx)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// handleJob is the workhorse of Neoq
|
||||
// it receives pending, periodic, and retry job ids asynchronously
|
||||
// 1. handleJob first creates a transactions inside of which a row lock is acquired for the job to be processed.
|
||||
// 2. handleJob secondly calls the handler on the job, and finally updates the job's status
|
||||
func (w pgWorker) handleJob(ctx context.Context, jobID int64, handler Handler) (err error) {
|
||||
var tx pgx.Tx
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
log.Println("failed to acquire database connection to process jobs")
|
||||
return
|
||||
}
|
||||
defer conn.Release()
|
||||
|
||||
tx, err = conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer tx.Rollback(ctx) // rollback has no effect if the transaction has been committed
|
||||
|
||||
ctxv := handlerCtxVars{tx: tx}
|
||||
var job Job
|
||||
job, err = w.getPendingJob(ctx, tx, jobID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
ctxv.job = &job
|
||||
ctx = withHandlerContext(ctx, ctxv)
|
||||
|
||||
// check if the job is being retried and increment retry count accordingly
|
||||
if job.Status != JobStatusNew {
|
||||
job.Retries = job.Retries + 1
|
||||
}
|
||||
|
||||
// execute the queue handler of this job
|
||||
handlerErr := w.execHandler(ctx, handler)
|
||||
err = w.updateJob(ctx, handlerErr)
|
||||
if err != nil {
|
||||
log.Println("error updating job status:", err)
|
||||
return
|
||||
}
|
||||
|
||||
err = tx.Commit(ctx)
|
||||
if err != nil {
|
||||
log.Printf("unable to commit transaction for job '%d': work may be duplicated if retried", job.ID)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// exechandler executes handler functions with a concrete time deadline
|
||||
func (w pgWorker) execHandler(ctx context.Context, handler Handler) (err error) {
|
||||
deadlineCtx, cancel := context.WithDeadline(ctx, time.Now().Add(handler.Deadline))
|
||||
defer cancel()
|
||||
|
||||
var done = make(chan bool)
|
||||
go func(ctx context.Context) {
|
||||
err = handler.Handle(ctx)
|
||||
done <- true
|
||||
}(ctx)
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
return
|
||||
case <-deadlineCtx.Done():
|
||||
err = fmt.Errorf("job exceeded its %s deadline", handler.Deadline)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// listen uses Postgres LISTEN to listen for jobs on a queue
|
||||
func (w pgWorker) listen(ctx context.Context, queue string) (c chan int64) {
|
||||
var err error
|
||||
c = make(chan int64)
|
||||
|
||||
_, err = w.listenConn.Exec(ctx, fmt.Sprintf("SET idle_in_transaction_session_timeout = '0'; LISTEN %s", queue))
|
||||
if err != nil {
|
||||
log.Println(err)
|
||||
return
|
||||
}
|
||||
|
||||
go func(ctx context.Context) {
|
||||
for {
|
||||
notification, waitErr := w.listenConn.WaitForNotification(ctx)
|
||||
if waitErr != nil {
|
||||
log.Println("failed to wait for notification:", waitErr)
|
||||
time.Sleep(1 * time.Second)
|
||||
continue
|
||||
}
|
||||
|
||||
var jobID int64
|
||||
if jobID, err = strconv.ParseInt(notification.Payload, 0, 64); err != nil {
|
||||
log.Println("unable to fetch job:", err)
|
||||
continue
|
||||
}
|
||||
|
||||
c <- jobID
|
||||
}
|
||||
}(ctx)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (w pgWorker) getPendingJob(ctx context.Context, tx pgx.Tx, jobID int64) (job Job, err error) {
|
||||
row, err := tx.Query(ctx, PendingJobQuery, jobID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var j *Job
|
||||
j, err = pgx.CollectOneRow(row, pgx.RowToAddrOfStructByName[Job])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return *j, err
|
||||
}
|
||||
|
||||
// calculateBackoff calculates tne number of seconds to back off before the next retry
|
||||
// this formula is unabashedly taken from Sidekiq because it is good
|
||||
func calculateBackoff(retryCount int) time.Time {
|
||||
p := int(math.Round(math.Pow(float64(retryCount), 4)))
|
||||
return time.Now().Add(time.Duration(p+15+randInt(30)*retryCount+1) * time.Second)
|
||||
}
|
||||
|
||||
func randInt(max int) int {
|
||||
rand.Seed(time.Now().UnixNano())
|
||||
return rand.Intn(max)
|
||||
}
|
||||
|
||||
func (w pgWorker) getPendingJobID(ctx context.Context, conn *pgxpool.Conn, queue string) (jobID int64, err error) {
|
||||
err = conn.QueryRow(ctx, PendingJobIDQuery, queue).Scan(&jobID)
|
||||
return
|
||||
}
|
||||
|
||||
// utility function for removing cancel functions from a slice cancel function pointers
|
||||
func remove(slice []context.CancelCauseFunc, i int) []context.CancelCauseFunc {
|
||||
return append(slice[:i], slice[i+1:]...)
|
||||
}
|
||||
82
neoq_test.go
Normal file
82
neoq_test.go
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
package neoq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestWorkerListenConn(t *testing.T) {
|
||||
const queue = "foobar"
|
||||
nq, err := New("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
jobRan := false
|
||||
numJobs := 1
|
||||
var done = make(chan bool, numJobs)
|
||||
handler := NewHandler(func(ctx context.Context) (err error) {
|
||||
j, err := JobFromContext(ctx)
|
||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||
done <- true
|
||||
return
|
||||
})
|
||||
handler = handler.
|
||||
WithOption(WithDeadline(time.Duration(200 * time.Millisecond))).
|
||||
WithOption(WithConcurrency(8))
|
||||
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
// Listen for jobs on the queue
|
||||
nq.Listen(queue, handler)
|
||||
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
|
||||
for i := 0; i < numJobs; i++ {
|
||||
jid, err := nq.Enqueue(Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": fmt.Sprintf("hello world: %d", 100),
|
||||
},
|
||||
})
|
||||
if err != nil || jid == -1 {
|
||||
t.Fatal("job was not enqueued. either it was duplicate or this error caused it:", err)
|
||||
}
|
||||
}
|
||||
|
||||
timeout := false
|
||||
doneCnt := 0
|
||||
for {
|
||||
select {
|
||||
case <-time.After(5 * time.Second):
|
||||
timeout = true
|
||||
err = errors.New("timed out waiting for job")
|
||||
break
|
||||
case <-done:
|
||||
doneCnt++
|
||||
}
|
||||
|
||||
if doneCnt >= numJobs {
|
||||
jobRan = true
|
||||
break
|
||||
}
|
||||
|
||||
if timeout {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Allow time for job status to be updated in the database
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
if !jobRan {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in a new issue