krz/gitbay

A CLI-first git forge. cli forge git self-hosted https://gitbay.org

Commit 6a187ab6f5

6a187ab6f5e829d81cdab54f846992be56cd78bc

parent: 0030cf8082

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-20 09:46 UTC 

push: drain the queue, started by gitbayd

Two-second ticker in the mailer's shape. A reap drops the device and
its queued rows cascade; a retry honours Retry-After when APNs gave
one. Log lines name the device id, never the token.

Ref #89
cmd/gitbayd/main.go +12
@@ -30,6 +30,7 @@ import (
3030 "gitbay.org/gitbay/internal/httpd"
3131 "gitbay.org/gitbay/internal/mirror"
3232 "gitbay.org/gitbay/internal/notify"
33 "gitbay.org/gitbay/internal/push"
3334 "gitbay.org/gitbay/internal/sshd"
3435 "gitbay.org/gitbay/internal/store"
3536 "gitbay.org/gitbay/internal/toolpath"
@@ -176,6 +177,17 @@ func serveCmd() *cobra.Command {
176177 if cfg.Mail.SMTPHost != "" {
177178 go notify.New(st, cfg, retryBase).Run(whCtx)
178179 }
180 if cfg.Push.Enabled {
181 p, err := push.New(st, cfg.Push, retryBase)
182 if err != nil {
183 // Config validation already parsed the key, so this
184 // is not a misconfiguration; fail loudly rather than
185 // running with a silent delivery route.
186 slog.Error("push: starting deliverer", "err", err)
187 } else {
188 go p.Run(whCtx)
189 }
190 }
179191 go mirror.New(st, cfg).Run(whCtx)
180192 if d := cfg.Registration.PendingExpiryDuration(); d > 0 {
181193 go reapPending(whCtx, st, d)
internal/push/push.go added +85
@@ -0,0 +1,85 @@
1package push
2
3import (
4 "context"
5 "log/slog"
6 "time"
7
8 "gitbay.org/gitbay/internal/config"
9 "gitbay.org/gitbay/internal/store"
10)
11
12// DefaultMaxAttempts matches the mailer's: a flaky APNs delays a
13// notification rather than losing it, up to a point.
14const DefaultMaxAttempts = 5
15
16type Deliverer struct {
17 St *store.Store
18 Cl *Client
19 RetryBase time.Duration
20 MaxAttempts int
21}
22
23func New(st *store.Store, cfg config.Push, retryBase time.Duration) (*Deliverer, error) {
24 cl, err := NewClient(cfg)
25 if err != nil {
26 return nil, err
27 }
28 return &Deliverer{St: st, Cl: cl, RetryBase: retryBase, MaxAttempts: DefaultMaxAttempts}, nil
29}
30
31// Run drains the push queue until ctx is done.
32func (d *Deliverer) Run(ctx context.Context) {
33 tick := time.NewTicker(2 * time.Second)
34 defer tick.Stop()
35 for {
36 select {
37 case <-ctx.Done():
38 return
39 case <-tick.C:
40 d.drain(ctx)
41 }
42 }
43}
44
45func (d *Deliverer) drain(ctx context.Context) {
46 due, err := d.St.DuePush(20)
47 if err != nil {
48 slog.Error("push: listing due", "err", err)
49 return
50 }
51 for _, q := range due {
52 res, after, sendErr := d.Cl.Send(ctx, q.Token, q.Title, q.Body, q.Path)
53 msg := ""
54 if sendErr != nil {
55 msg = sendErr.Error()
56 }
57 switch res {
58 case resultSent:
59 d.St.MarkPushSent(q.ID)
60 case resultReap:
61 // The queued rows cascade with the device.
62 if err := d.St.DeletePushDeviceByToken(q.Token); err != nil {
63 slog.Error("push: reaping device", "device", q.DeviceID, "err", err)
64 }
65 case resultRetry:
66 attempt := q.Attempts + 1
67 if attempt >= d.MaxAttempts {
68 d.St.MarkPushFailed(q.ID, msg, nil)
69 // The device id, never the token.
70 slog.Warn("push dead-lettered",
71 "push", q.ID, "device", q.DeviceID, "attempts", attempt, "err", msg)
72 continue
73 }
74 wait := after
75 if wait == 0 {
76 wait = d.RetryBase << (attempt - 1)
77 }
78 next := time.Now().Add(wait)
79 d.St.MarkPushFailed(q.ID, msg, &next)
80 default: // resultDead
81 d.St.MarkPushFailed(q.ID, msg, nil)
82 slog.Warn("push rejected", "push", q.ID, "device", q.DeviceID, "err", msg)
83 }
84 }
85}
internal/push/push_test.go added +112
@@ -0,0 +1,112 @@
1package push
2
3import (
4 "context"
5 "net/http"
6 "testing"
7 "time"
8
9 "gitbay.org/gitbay/internal/store"
10)
11
12// testStoreWithQueuedPush opens an in-memory store, creates a user with
13// push enabled, registers one device and enqueues one push for it.
14func testStoreWithQueuedPush(t *testing.T, token string) *store.Store {
15 t.Helper()
16 st, err := store.Open(":memory:")
17 if err != nil {
18 t.Fatal(err)
19 }
20 t.Cleanup(func() { st.Close() })
21 if err := st.MigrateUp(); err != nil {
22 t.Fatal(err)
23 }
24 uid, err := st.CreateUser("alice", false)
25 if err != nil {
26 t.Fatal(err)
27 }
28 if err := st.SetPushEnabled(uid, true); err != nil {
29 t.Fatal(err)
30 }
31 if _, err := st.AddPushDevice(uid, token, "iphone"); err != nil {
32 t.Fatal(err)
33 }
34 if err := st.EnqueuePush(uid, "krz/gitbay", "cmc opened issue #12", "krz/gitbay/issues/12"); err != nil {
35 t.Fatal(err)
36 }
37 return st
38}
39
40// countPushDevices counts the test user's own devices. testStoreWithQueuedPush
41// always creates "alice", so looking her up here keeps the helper's signature
42// matching the brief's test code, which passes no user id.
43func countPushDevices(t *testing.T, st *store.Store) int {
44 t.Helper()
45 u, err := st.UserByUsername("alice")
46 if err != nil {
47 t.Fatal(err)
48 }
49 devices, err := st.PushDevices(u.ID)
50 if err != nil {
51 t.Fatal(err)
52 }
53 return len(devices)
54}
55
56func TestDrainSendsAndMarks(t *testing.T) {
57 var hits int
58 c, _ := fakeAPNs(t, func(w http.ResponseWriter, r *http.Request) {
59 hits++
60 w.WriteHeader(200)
61 })
62 st := testStoreWithQueuedPush(t, "tok-a")
63 d := &Deliverer{St: st, Cl: c, RetryBase: time.Millisecond, MaxAttempts: 5}
64
65 d.drain(context.Background())
66
67 if hits != 1 {
68 t.Fatalf("sent %d times, want 1", hits)
69 }
70 if due, _ := st.DuePush(20); len(due) != 0 {
71 t.Fatalf("row still due after a 200")
72 }
73}
74
75func TestDrainReapsADeadToken(t *testing.T) {
76 c, _ := fakeAPNs(t, func(w http.ResponseWriter, r *http.Request) {
77 w.WriteHeader(410)
78 w.Write([]byte(`{"reason":"Unregistered"}`))
79 })
80 st := testStoreWithQueuedPush(t, "tok-a")
81 d := &Deliverer{St: st, Cl: c, RetryBase: time.Millisecond, MaxAttempts: 5}
82
83 d.drain(context.Background())
84
85 if due, _ := st.DuePush(20); len(due) != 0 {
86 t.Fatalf("queue survived the reap")
87 }
88 // The device is gone, not merely its queue row.
89 if n := countPushDevices(t, st); n != 0 {
90 t.Fatalf("%d devices left after 410", n)
91 }
92}
93
94func TestDrainBacksOffThenDeadLetters(t *testing.T) {
95 c, _ := fakeAPNs(t, func(w http.ResponseWriter, r *http.Request) {
96 w.WriteHeader(503)
97 })
98 st := testStoreWithQueuedPush(t, "tok-a")
99 d := &Deliverer{St: st, Cl: c, RetryBase: time.Nanosecond, MaxAttempts: 3}
100
101 // Three passes: two back off, the third gives up.
102 for i := 0; i < 3; i++ {
103 d.drain(context.Background())
104 }
105 if due, _ := st.DuePush(20); len(due) != 0 {
106 t.Fatalf("row still due after MaxAttempts")
107 }
108 // A transient failure must not take the device with it.
109 if n := countPushDevices(t, st); n != 1 {
110 t.Fatalf("device reaped on a 503")
111 }
112}