Skip to content

Commit 567b24e

Browse files
MohamedAklamaashMohamedAklamaash
committed
add tests, examples, ci and docs
1 parent 4125205 commit 567b24e

10 files changed

Lines changed: 753 additions & 0 deletions

File tree

‎.github/workflows/ci.yml‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
name: ci
2+
3+
on:
4+
push:
5+
branches: [main]
6+
pull_request:
7+
8+
jobs:
9+
test:
10+
runs-on: ubuntu-latest
11+
services:
12+
redis:
13+
image: redis:7-alpine
14+
ports:
15+
- 6379:6379
16+
options: >-
17+
--health-cmd "redis-cli ping"
18+
--health-interval 5s
19+
--health-timeout 3s
20+
--health-retries 5
21+
steps:
22+
- uses: actions/checkout@v4
23+
- uses: actions/setup-go@v5
24+
with:
25+
go-version: "1.24"
26+
- run: go vet ./...
27+
- run: go build ./...
28+
- run: go test -race ./...
29+
env:
30+
RELAY_REDIS_ADDR: 127.0.0.1:6379
31+
32+
lint:
33+
runs-on: ubuntu-latest
34+
steps:
35+
- uses: actions/checkout@v4
36+
- uses: actions/setup-go@v5
37+
with:
38+
go-version: "1.24"
39+
- name: gofmt
40+
run: test -z "$(gofmt -l .)"

‎CHANGELOG.md‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
# Changelog
2+
3+
All notable changes to this project are documented here. The format follows
4+
[Keep a Changelog](https://keepachangelog.com/) and the project adheres to
5+
semantic versioning.
6+
7+
## [Unreleased]
8+
9+
### Added
10+
- Redis-backed client, worker server, and cron scheduler
11+
- At-least-once delivery with visibility-timeout crash recovery
12+
- Immediate, delayed, and recurring cron jobs
13+
- Automatic retries with exponential backoff and dead-letter archive
14+
- Task deduplication via `Unique` and explicit `TaskID`
15+
- Weighted priority queues with pause/resume
16+
- Per-task timeouts, deadlines, and remote cancellation
17+
- Prometheus metrics
18+
- `relay` CLI for queue inspection and management

‎README.md‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
# relay
2+
3+
A simple, reliable, distributed job scheduler for Go, backed by Redis.
4+
5+
Enqueue background jobs from anywhere, process them across a fleet of workers, and schedule recurring work with cron. Jobs survive worker crashes and are retried automatically.
6+
7+
## Features
8+
9+
- At-least-once delivery with automatic crash recovery (visibility-timeout leases)
10+
- Immediate, delayed (`ProcessIn` / `ProcessAt`), and recurring **cron** jobs
11+
- Automatic retries with exponential backoff and a dead-letter archive
12+
- Optional task deduplication (`Unique`) and explicit task IDs
13+
- Weighted priority queues with pause/resume
14+
- Per-task timeouts, deadlines, and remote cancellation
15+
- Prometheus metrics out of the box
16+
- A CLI to inspect and manage queues
17+
18+
## Install
19+
20+
```sh
21+
go get github.com/MohamedAklamaash/relay
22+
```
23+
24+
Requires Go 1.24+ and a Redis server.
25+
26+
## Quickstart
27+
28+
Enqueue a task:
29+
30+
```go
31+
client := relay.NewClient(relay.RedisClientOpt{Addr: "127.0.0.1:6379"})
32+
defer client.Close()
33+
34+
client.Enqueue(relay.NewTask("email:send", []byte(`{"to":"a@b.com"}`)))
35+
client.Enqueue(relay.NewTask("report:build", nil), relay.ProcessIn(10*time.Minute))
36+
```
37+
38+
Process tasks:
39+
40+
```go
41+
srv := relay.NewServer(relay.RedisClientOpt{Addr: "127.0.0.1:6379"}, relay.Config{
42+
Concurrency: 10,
43+
Queues: map[string]int{"critical": 6, "default": 3, "low": 1},
44+
})
45+
46+
mux := relay.NewServeMux()
47+
mux.HandleFunc("email:send", func(ctx context.Context, t *relay.Task) error {
48+
return sendEmail(t.Payload())
49+
})
50+
51+
srv.Run(mux)
52+
```
53+
54+
Schedule recurring work:
55+
56+
```go
57+
scheduler := relay.NewScheduler(relay.RedisClientOpt{Addr: "127.0.0.1:6379"}, nil)
58+
scheduler.Register("0 * * * *", relay.NewTask("report:build", nil))
59+
scheduler.Run()
60+
```
61+
62+
## Enqueue options
63+
64+
| Option | Description |
65+
| --- | --- |
66+
| `Queue(name)` | Target queue (default `"default"`) |
67+
| `ProcessIn(d)` / `ProcessAt(t)` | Delay execution |
68+
| `MaxRetry(n)` | Retry attempts before archiving (default 25) |
69+
| `Timeout(d)` / `Deadline(t)` | Cancel the handler context after a limit |
70+
| `Unique(ttl)` | Reject duplicate enqueues within the window |
71+
| `TaskID(id)` | Set an explicit ID; conflicting IDs are rejected |
72+
| `Retention(d)` | Keep completed task data for inspection |
73+
74+
## Delivery semantics
75+
76+
`relay` guarantees **at-least-once** delivery. A task may run more than once (for
77+
example when a worker crashes after finishing but before acknowledging), so your
78+
handlers must be **idempotent**. `Unique` deduplicates *enqueues*, not executions.
79+
80+
Return `relay.SkipRetry` from a handler to archive a task immediately instead of
81+
retrying.
82+
83+
## Metrics
84+
85+
`relay` exposes Prometheus metrics (processed, failed, retried, archived,
86+
in-progress, and processing duration). Mount the handler on your own server:
87+
88+
```go
89+
http.Handle("/metrics", relay.MetricsHandler())
90+
go http.ListenAndServe(":2112", nil)
91+
```
92+
93+
## Cancellation
94+
95+
Cancel a running task from anywhere; the worker's handler context is cancelled:
96+
97+
```go
98+
insp := relay.NewInspector(relay.RedisClientOpt{Addr: "127.0.0.1:6379"})
99+
insp.CancelTask(taskID)
100+
```
101+
102+
## CLI
103+
104+
```sh
105+
go install github.com/MohamedAklamaash/relay/cmd/relay@latest
106+
107+
relay stats
108+
relay ls default retry
109+
relay run default <task-id>
110+
relay rm default <task-id>
111+
relay cancel <task-id>
112+
relay pause low
113+
```
114+
115+
## Running locally
116+
117+
```sh
118+
docker compose up -d redis
119+
go run ./examples/worker
120+
go run ./examples/producer
121+
go run ./examples/cron
122+
```
123+
124+
## License
125+
126+
MIT

‎docker-compose.yml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
services:
2+
redis:
3+
image: redis:7-alpine
4+
ports:
5+
- "6379:6379"
6+
command: ["redis-server", "--appendonly", "yes"]
7+
volumes:
8+
- relay-redis-data:/data
9+
10+
volumes:
11+
relay-redis-data:

‎examples/cron/main.go‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package main
2+
3+
import (
4+
"log"
5+
6+
"github.com/MohamedAklamaash/relay"
7+
)
8+
9+
func main() {
10+
scheduler := relay.NewScheduler(relay.RedisClientOpt{Addr: "127.0.0.1:6379"}, nil)
11+
12+
if _, err := scheduler.Register("*/1 * * * *", relay.NewTask("report:build", []byte(`{"kind":"minutely"}`))); err != nil {
13+
log.Fatal(err)
14+
}
15+
if _, err := scheduler.Register("0 * * * *", relay.NewTask("email:send", []byte(`{"to":"ops@example.com","subject":"hourly"}`)), relay.Queue("critical")); err != nil {
16+
log.Fatal(err)
17+
}
18+
19+
if err := scheduler.Run(); err != nil {
20+
log.Fatal(err)
21+
}
22+
}

‎examples/producer/main.go‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
package main
2+
3+
import (
4+
"encoding/json"
5+
"log"
6+
"time"
7+
8+
"github.com/MohamedAklamaash/relay"
9+
)
10+
11+
type EmailPayload struct {
12+
To string `json:"to"`
13+
Subject string `json:"subject"`
14+
}
15+
16+
func main() {
17+
client := relay.NewClient(relay.RedisClientOpt{Addr: "127.0.0.1:6379"})
18+
defer client.Close()
19+
20+
payload, _ := json.Marshal(EmailPayload{To: "user@example.com", Subject: "welcome"})
21+
22+
now, err := client.Enqueue(relay.NewTask("email:send", payload), relay.Queue("critical"))
23+
if err != nil {
24+
log.Fatal(err)
25+
}
26+
log.Printf("enqueued immediate task %s", now.ID)
27+
28+
later, err := client.Enqueue(
29+
relay.NewTask("report:build", []byte(`{"month":"june"}`)),
30+
relay.ProcessIn(10*time.Second),
31+
relay.MaxRetry(5),
32+
)
33+
if err != nil {
34+
log.Fatal(err)
35+
}
36+
log.Printf("enqueued delayed task %s for %s", later.ID, later.NextProcessAt.Format(time.RFC3339))
37+
38+
_, err = client.Enqueue(
39+
relay.NewTask("email:send", payload),
40+
relay.Queue("critical"),
41+
relay.Unique(time.Minute),
42+
)
43+
if err != nil {
44+
log.Printf("second unique enqueue rejected as expected: %v", err)
45+
}
46+
}

‎examples/worker/main.go‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"log"
7+
"net/http"
8+
9+
"github.com/MohamedAklamaash/relay"
10+
)
11+
12+
type EmailPayload struct {
13+
To string `json:"to"`
14+
Subject string `json:"subject"`
15+
}
16+
17+
func main() {
18+
srv := relay.NewServer(relay.RedisClientOpt{Addr: "127.0.0.1:6379"}, relay.Config{
19+
Concurrency: 10,
20+
Queues: map[string]int{"critical": 6, "default": 3, "low": 1},
21+
})
22+
23+
mux := relay.NewServeMux()
24+
mux.HandleFunc("email:send", handleEmail)
25+
mux.HandleFunc("report:build", handleReport)
26+
27+
http.Handle("/metrics", relay.MetricsHandler())
28+
go http.ListenAndServe(":2112", nil)
29+
30+
if err := srv.Run(mux); err != nil {
31+
log.Fatal(err)
32+
}
33+
}
34+
35+
func handleEmail(ctx context.Context, t *relay.Task) error {
36+
var p EmailPayload
37+
if err := json.Unmarshal(t.Payload(), &p); err != nil {
38+
return err
39+
}
40+
log.Printf("sending email to %s: %s", p.To, p.Subject)
41+
return nil
42+
}
43+
44+
func handleReport(ctx context.Context, t *relay.Task) error {
45+
log.Printf("building report: %s", string(t.Payload()))
46+
return nil
47+
}

0 commit comments

Comments
 (0)