feat(apps): per-endpoint webhook delivery state, a per-install rate cap, and a 24 h drop (TASK-3408, SPEC-6 U10c) (#1783)

* feat(apps): per-endpoint webhook delivery state, a per-install rate cap, and a 24 h drop (TASK-3408, SPEC-6 U10c)

webhook_deliveries records each app hook's decision per outbox event. A
terminal row (delivered, permanent, refused, skipped, dropped) is never
retried, so a retry goes only to app endpoints still owed, and an event
an app was refused once (held, disabled, not visible) stays refused,
which closes U10b's deliver_from precision and clock-skew residuals for
any event refused once. A pending row (transient, deferred, rate_limited)
keeps the event owed.

Owner hooks record only "delivered", used only to skip re-sending an
event an owner endpoint already received while it stays owed to an app
(lead ruling, day 87). Owner failure, retry and ack are unchanged; full
owner per-endpoint state is TASK-3409.

A per-install cap of 600 a minute (burst 60, per instance) defers the
excess without charging an attempt. A delivery still owed 24 h after its
event is dropped, counted on the hook and shown to the owner as
undelivered_dropped.

Migrations 124 / pg 098.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* fix(apps): rate cap counts every attempt and only subscribed targets; limiter pruned; bulk skips recorded (codex r1 on U10c, TASK-3408)

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* fix(apps): only owed deliveries are dropped; the rate cap is asked after admission; the ledger counts requests sent (codex r2 on U10c, TASK-3408)

Admission's checks become one function run under the install lock for
admission and as a plain read for AppDeliveryRefusal, so a 24 h-old
delivery admission would refuse is recorded refused, not dropped and
counted. The rate gate moves after admission, so a refused attempt spends
nothing. DeliverAppEvent returns the requests it sent, which the ledger
adds.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* fix(apps): a request that failed before sending is not counted (codex r3 on U10c, TASK-3408)

appfetch.Poster wraps dial and TLS-handshake failures in ErrNotSent; the
dispatcher counts a request as sent only when the failure is neither that
nor a policy refusal.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* fix(apps): the drop decision runs under the install lock; a write failing before its first byte is unsent (codex r4 on U10c, TASK-3408)

DropOwedDelivery takes the install row and runs admission's checks in
the same transaction that records the outcome, so a disable or upgrade
cannot land between the check and the drop. The Poster counts bytes
written through its connection; a write that fails at zero bytes is
ErrNotSent.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* fix(apps): a late delivery replaces a drop and takes it off the count (codex r5 on U10c, TASK-3408)

A request admitted before another drainer dropped its event (the first
drainer's outbox claim expired mid-request) did reach the app. RecordDelivery
now reads the row under its own transaction: delivered replaces dropped and
decrements dropped_count; every other terminal decision still stands.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* fix(apps): RecordDelivery inserts before it locks, so a concurrent drop is seen (codex r6 on U10c, TASK-3408)

SELECT ... FOR UPDATE locks nothing when the row does not exist, so a drop
committing between the read and the upsert was overwritten with the count
left raised. RecordDelivery now inserts with ON CONFLICT DO NOTHING (which
waits for a concurrent inserter), and only when the row already existed
locks and reads it before deciding. A deterministic test drives a drop
between the two statements; run against the previous code on Postgres it
ends delivered with dropped_count 1.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* test(apps): the drop/record race test waits for the drop to reach its write (codex r7 on U10c, TASK-3408)

Against the previous RecordDelivery on Postgres it fails 3 of 3
(delivered, dropped_count 1); with the fix it passes on both dialects.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS
This commit is contained in:
xarmian
2026-10-05 00:02:49 -04:00
committed by GitHub
parent b0dc7d51e5
commit be54575954
16 changed files with 1316 additions and 75 deletions
+34 -5
View File
@@ -45,6 +45,12 @@ type PrivateOrigin struct {
// address or size. The wrapped message says which.
var ErrRefused = errors.New("app fetch refused")
// ErrNotSent marks a Poster.Post failure that happened before any byte of
// the request was written (dial, TLS handshake), so a delivery ledger does
// not count a request the endpoint never received (codex r3 on U10c). A
// policy refusal (ErrRefused) is also unsent.
var ErrNotSent = errors.New("request not sent")
func refused(format string, args ...any) error {
return fmt.Errorf("%w: %s", ErrRefused, fmt.Sprintf(format, args...))
}
@@ -335,7 +341,7 @@ func (p *Poster) Post(ctx context.Context, rawURL string, body []byte, header ht
// itself, so nothing is resolved between the screen and the connect.
raw, err := p.f.dial(context.WithValue(ctx, originKey{}, origin), p.f.dialer, "tcp", net.JoinHostPort(u.Hostname(), port))
if err != nil {
return 0, err
return 0, fmt.Errorf("%w: %w", ErrNotSent, err)
}
// The system roots unless a test supplies its own; the hostname is
// verified against the certificate (ServerName); never InsecureSkipVerify.
@@ -363,7 +369,7 @@ func (p *Poster) Post(ctx context.Context, rawURL string, body []byte, header ht
err = conn.HandshakeContext(hctx)
hcancel()
if err != nil {
return 0, err
return 0, fmt.Errorf("%w: %w", ErrNotSent, err)
}
req, err := http.NewRequest(http.MethodPost, u.String(), bytes.NewReader(body))
if err != nil {
@@ -375,12 +381,16 @@ func (p *Poster) Post(ctx context.Context, rawURL string, body []byte, header ht
}
}
req.Close = true // one request per connection
bw := bufio.NewWriter(conn)
// Count what reached the connection: a write that fails before its
// first byte (a reset or the deadline right after the handshake) sent
// nothing (codex r4 on U10c).
cw := &countingWriter{w: conn}
bw := bufio.NewWriter(cw)
if err := req.Write(bw); err != nil {
return 0, err
return 0, unsentIfNothingWritten(cw, err)
}
if err := bw.Flush(); err != nil {
return 0, err
return 0, unsentIfNothingWritten(cw, err)
}
// Interim 1xx answers (100 Continue, 103 Early Hints) precede the final
// one and are skipped, as net/http's Transport does (codex r4 on U10b);
@@ -422,3 +432,22 @@ func underPath(u *url.URL, prefix string) bool {
}
return u.Path == prefix || strings.HasPrefix(u.Path, prefix+"/")
}
// countingWriter counts the bytes its writer accepted.
type countingWriter struct {
w io.Writer
n int64
}
func (c *countingWriter) Write(p []byte) (int, error) {
n, err := c.w.Write(p)
c.n += int64(n)
return n, err
}
func unsentIfNothingWritten(cw *countingWriter, err error) error {
if cw.n == 0 {
return fmt.Errorf("%w: %w", ErrNotSent, err)
}
return err
}
+50
View File
@@ -238,3 +238,53 @@ func TestPoster_EntryPinsAndPathsStayPaired(t *testing.T) {
t.Fatalf("/a reached /b's address: %d %v", st, err)
}
}
// codex r3 on U10c: failures before any request byte (dial, handshake) are
// ErrNotSent, so a ledger does not count them as requests.
func TestPoster_PreSendFailuresAreNotSent(t *testing.T) {
srv, _ := tlsServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(204) }))
entry := []PrivateOrigin{{Origin: srv.URL, Allowed: []string{"127.0.0.1"}, Webhook: true}}
// Handshake: the system roots do not trust the test server.
p, err := NewPoster(entry, 5*time.Second, nil)
if err != nil {
t.Fatal(err)
}
if _, err := p.Post(context.Background(), srv.URL+"/hooks", nil, http.Header{}); !errors.Is(err, ErrNotSent) {
t.Fatalf("handshake failure: %v, want ErrNotSent", err)
}
// Dial: nothing listens there any more.
closedURL := srv.URL
srv.Close()
if _, err := p.Post(context.Background(), closedURL+"/hooks", nil, http.Header{}); !errors.Is(err, ErrNotSent) {
t.Fatalf("dial failure: %v, want ErrNotSent", err)
}
}
type failAfter struct{ ok int }
func (f *failAfter) Write(p []byte) (int, error) {
if f.ok <= 0 {
return 0, errors.New("connection reset")
}
n := min(len(p), f.ok)
f.ok -= n
if n < len(p) {
return n, errors.New("connection reset")
}
return n, nil
}
// codex r4 on U10c: a write that fails before its first byte is unsent; one
// that fails part-way is not.
func TestPoster_ZeroByteWriteFailureIsUnsent(t *testing.T) {
none := &countingWriter{w: &failAfter{}}
_, err := none.Write([]byte("POST /"))
if !errors.Is(unsentIfNothingWritten(none, err), ErrNotSent) {
t.Fatal("a write failing at its first byte was not ErrNotSent")
}
some := &countingWriter{w: &failAfter{ok: 3}}
_, err = some.Write([]byte("POST /"))
if errors.Is(unsentIfNothingWritten(some, err), ErrNotSent) {
t.Fatal("a write failing after 3 bytes was reported unsent")
}
}
+163 -6
View File
@@ -8,6 +8,8 @@ import (
"sync"
"time"
"golang.org/x/time/rate"
"github.com/PerpetualSoftware/pad/internal/appfetch"
"github.com/PerpetualSoftware/pad/internal/store"
"github.com/PerpetualSoftware/pad/internal/webhooks"
@@ -82,7 +84,19 @@ func (s *Server) deliverAppHooks(unit outboxDelivery) (owed bool, err error) {
if err != nil {
return false, err
}
for range hooks {
for _, h := range hooks {
// Recorded, so an owner hook's retry of this unit does not count
// it again (codex r1 on U10c).
st, err := s.store.DeliveryStatus(unit.eventID, h.WebhookID)
if err != nil {
return false, err
}
if store.DeliveryTerminal(st) {
continue
}
if err := s.store.RecordDelivery(unit.eventID, h.WebhookID, store.DeliverySkipped, "bulk operation", 0); err != nil {
return false, err
}
s.countAppDelivery("skipped_bulk")
}
return false, nil
@@ -94,10 +108,29 @@ func (s *Server) deliverAppHooks(unit outboxDelivery) (owed bool, err error) {
if err != nil || len(targets) == 0 {
return false, err
}
// Per-endpoint state (U10c): an endpoint whose row is terminal is done
// with this event and never retried.
var live []store.AppHookTarget
for _, t := range targets {
st, err := s.store.DeliveryStatus(unit.eventID, t.WebhookID)
if err != nil {
return false, err
}
if !store.DeliveryTerminal(st) {
live = append(live, t)
}
}
if len(live) == 0 {
return false, nil
}
body, collectionID, err := store.BuildAppEventDTO(unit.eventType, unit.eventID, unit.occurredAt, unit.payload)
if errors.Is(err, store.ErrNoAppProjection) {
// Skipped and counted, never filled in from live state (§5).
for range targets {
// Skipped and counted, never filled in from live state (§5), and
// recorded so a retry for another endpoint does not count it again.
for _, t := range live {
if err := s.store.RecordDelivery(unit.eventID, t.WebhookID, store.DeliverySkipped, "no app projection", 0); err != nil {
return false, err
}
s.countAppDelivery("skipped_no_projection")
}
slog.Warn("app webhook: event has no app projection; skipped", "event_id", unit.eventID, "event", unit.eventType)
@@ -111,14 +144,138 @@ func (s *Server) deliverAppHooks(unit outboxDelivery) (owed bool, err error) {
return false, err
}
adm := appAdmitter{st: s.store}
for _, t := range targets {
res := s.webhooks.DeliverAppEvent(adm, poster, webhooks.AppDelivery{
expired := appDeliveryExpired(unit.occurredAt, time.Now())
for _, t := range live {
// Only a hook subscribed to this event FOR THIS COLLECTION is a
// target: anything else would be refused at admission, and must not
// spend the install's rate budget or be dropped (codex r1 on U10c).
if !store.AppHookSubscribes(t.Events, unit.eventType, collectionID) {
continue
}
if expired {
// Only a delivery the app was OWED is dropped and counted, and
// the decision is made under the install lock, as admission's is
// (codex r2/r4 on U10c): one admission would refuse is recorded
// refused.
outcome, err := s.store.DropOwedDelivery(unit.eventID, t.WebhookID, unit.eventType, collectionID, unit.occurredAt, "undelivered after "+appWebhookDropAfter.String())
if err != nil {
return false, err
}
switch {
case outcome == "dropped":
s.countAppDelivery("dropped")
case outcome != "":
s.countAppDelivery("refused")
}
continue
}
installID := t.InstallID
res, sent := s.webhooks.DeliverAppEvent(adm, poster, webhooks.AppDelivery{
WebhookID: t.WebhookID, Event: unit.eventType, CollectionID: collectionID, OccurredAt: unit.occurredAt, Body: body,
// The cap is per ATTEMPT: a retry spends a token too.
RateGate: func() bool { return s.appRate.allow(installID) },
})
s.countAppDelivery(res.String())
if res == webhooks.AppTransient || res == webhooks.AppDeferred {
status := appDeliveryStatus(res)
if err := s.store.RecordDelivery(unit.eventID, t.WebhookID, status, "", sent); err != nil {
return false, err
}
if !store.DeliveryTerminal(status) {
owed = true
}
}
return owed, nil
}
// appWebhookDropAfter: an app delivery still undelivered this long after its
// event occurred is dropped and counted on the hook (dropped_count, shown to
// the owner) rather than retried. 24 h (lead ruling, day 86): long enough to
// ride out an app's ordinary outage or deploy, short enough that an app
// receiving a day-old change learns it should resync instead. Well inside
// the outbox's 7-day undispatched retention.
const appWebhookDropAfter = 24 * time.Hour
func appDeliveryExpired(occurredAt string, now time.Time) bool {
t, err := time.Parse(time.RFC3339, occurredAt)
if err != nil {
return false
}
return now.Sub(t) > appWebhookDropAfter
}
// appDeliveryStatus maps an attempt's result to its recorded status.
func appDeliveryStatus(r webhooks.AppResult) string {
switch r {
case webhooks.AppDelivered:
return store.DeliveryDelivered
case webhooks.AppPermanent:
return store.DeliveryPermanent
case webhooks.AppRefused:
return store.DeliveryRefused
case webhooks.AppDeferred:
return store.DeliveryDeferred
case webhooks.AppRateLimited:
// Deferred, not dropped: the event stays owed and a later pass
// tries again.
return store.DeliveryRateLimited
default:
return store.DeliveryTransient
}
}
// The per-install delivery rate cap (lead ruling, day 86): 600 a minute with
// a burst of 60. One busy workspace with an installed app can emit faster
// than any app needs to hear, and the drain delivers synchronously, so an
// uncapped app would take the drain's whole pass. Excess is DEFERRED (the
// event stays owed and goes on a later pass), never dropped. PER INSTANCE:
// the bucket lives in this process, so N instances allow up to N times the
// rate; acceptable for v1, where Pad runs single-instance (CLAUDE.md, collab).
const (
appWebhookRatePerMinute = 600
appWebhookRateBurst = 60
)
type appRateLimiter struct {
mu sync.Mutex
by map[string]*rate.Limiter
// limit and burst are the constants above; fields so a test can tighten
// them.
limit rate.Limit
burst int
}
func (l *appRateLimiter) allow(installID string) bool {
l.mu.Lock()
defer l.mu.Unlock()
if l.by == nil {
l.by = map[string]*rate.Limiter{}
if l.limit == 0 {
l.limit, l.burst = rate.Limit(appWebhookRatePerMinute)/60, appWebhookRateBurst
}
}
lim, ok := l.by[installID]
if !ok {
if len(l.by) >= appRateLimiterPruneAt {
l.pruneIdleLocked()
}
lim = rate.NewLimiter(l.limit, l.burst)
l.by[installID] = lim
}
return lim.Allow()
}
// appRateLimiterPruneAt bounds the limiter map (codex r1 on U10c): past it,
// idle entries are dropped before a new one is added.
const appRateLimiterPruneAt = 1024
// pruneIdleLocked drops limiters whose bucket is full. A full bucket is
// exactly what a fresh limiter starts with, so dropping one changes no
// answer: an uninstalled or quiet install's entry goes, a busy one stays.
func (l *appRateLimiter) pruneIdleLocked() {
now := time.Now()
for id, lim := range l.by {
if lim.TokensAt(now) >= float64(l.burst) {
delete(l.by, id)
}
}
}
+3 -1
View File
@@ -56,6 +56,8 @@ type appWebhookView struct {
URL string `json:"url"`
Status string `json:"status"`
Notice string `json:"notice,omitempty"`
// UndeliveredDropped counts deliveries dropped undelivered after 24 h.
UndeliveredDropped int64 `json:"undelivered_dropped"`
}
func (s *Server) appWebhookView(installID string) (*appWebhookView, error) {
@@ -63,7 +65,7 @@ func (s *Server) appWebhookView(installID string) (*appWebhookView, error) {
if err != nil || st == nil {
return nil, err
}
v := &appWebhookView{URL: st.URL, Status: st.Status}
v := &appWebhookView{URL: st.URL, Status: st.Status, UndeliveredDropped: st.Dropped}
if st.Status == store.AppWebhookAwaitingSecret {
v.Notice = appWebhookHeldNotice
}
+356
View File
@@ -0,0 +1,356 @@
package server
import (
"fmt"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
dto "github.com/prometheus/client_model/go"
"golang.org/x/time/rate"
"github.com/PerpetualSoftware/pad/internal/metrics"
"github.com/PerpetualSoftware/pad/internal/models"
"github.com/PerpetualSoftware/pad/internal/store"
)
// TASK-3408 (U10c): per-endpoint delivery state, the owner no-duplicate
// ledger, the per-install rate cap and the 24 h drop.
// ownerSink is an owner webhook endpoint answering status, counting hits.
func ownerSink(t *testing.T, e *deliveryEnv, status *atomic.Int32) *atomic.Int32 {
t.Helper()
var hits atomic.Int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits.Add(1)
w.WriteHeader(int(status.Load()))
}))
t.Cleanup(srv.Close)
if _, err := e.srv.store.CreateWebhook(e.wsID, models.WebhookCreate{URL: srv.URL, Events: `["*"]`}); err != nil {
t.Fatal(err)
}
return &hits
}
// An app held back by its rate cap for several passes causes exactly ONE
// delivery to an owner hook that succeeded (lead ruling, day 87).
func TestAppDelivery_RateLimitedAppDoesNotDuplicateOwnerDeliveries(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
var ok atomic.Int32
ok.Store(http.StatusNoContent)
owner := ownerSink(t, e, &ok)
e.srv.appRate.mu.Lock()
e.srv.appRate.by, e.srv.appRate.limit, e.srv.appRate.burst = map[string]*rate.Limiter{}, 0, 0 // never allows
e.srv.appRate.mu.Unlock()
e.item(t, e.companion.ID, "Busy app")
for i := 0; i < 3; i++ {
e.tick(t)
}
if n := owner.Load(); n != 1 {
t.Fatalf("owner endpoint received %d deliveries while the app was rate-limited, want 1", n)
}
if got := e.hooksAt("/hooks"); len(got) != 0 {
t.Fatalf("a rate-limited app received %d deliveries", len(got))
}
if e.pendingOutbox(t) == 0 {
t.Fatal("a rate-limited app delivery did not keep the event owed")
}
// Deferred, not attempted: three passes charge no attempt.
var status string
var attempts int
if err := e.srv.store.DB().QueryRow(`SELECT d.status, d.attempts FROM webhook_deliveries d JOIN webhooks w ON w.id = d.webhook_id WHERE w.app_install_id = ?`, e.installID).Scan(&status, &attempts); err != nil {
t.Fatal(err)
}
if status != "rate_limited" || attempts != 0 {
t.Fatalf("app row %s/%d, want rate_limited/0", status, attempts)
}
// The cap lifts: the app gets it once, the owner still once, and the
// event is acked.
e.srv.appRate.mu.Lock()
e.srv.appRate.by, e.srv.appRate.limit, e.srv.appRate.burst = map[string]*rate.Limiter{}, rate.Inf, 1
e.srv.appRate.mu.Unlock()
e.tick(t)
e.tick(t)
if got := e.hooksAt("/hooks"); len(got) != 1 {
t.Fatalf("%d app deliveries after the cap lifted, want 1", len(got))
}
if n := owner.Load(); n != 1 {
t.Fatalf("owner endpoint received %d deliveries, want 1", n)
}
if n := e.pendingOutbox(t); n != 0 {
t.Fatalf("%d events still owed", n)
}
}
// An owner hook that FAILED is retried as today, while an app endpoint that
// already received the event is not sent it again.
func TestAppDelivery_FailingOwnerRetriedAppNotResent(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
var down atomic.Int32
down.Store(http.StatusServiceUnavailable)
owner := ownerSink(t, e, &down)
e.item(t, e.companion.ID, "Owner down")
e.tick(t)
first := owner.Load()
e.tick(t)
if owner.Load() <= first || first == 0 {
t.Fatalf("a failing owner hook was not retried (%d then %d hits)", first, owner.Load())
}
if got := e.hooksAt("/hooks"); len(got) != 1 {
t.Fatalf("the app received %d deliveries across the owner's retries, want 1", len(got))
}
}
// An app endpoint that failed transiently is retried, and once delivered is
// done.
func TestAppDelivery_TransientAppRetriedUntilDelivered(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
e.status["/hooks"] = http.StatusServiceUnavailable
e.item(t, e.companion.ID, "Flaky app")
e.tick(t)
if e.pendingOutbox(t) == 0 {
t.Fatal("a transient app failure acked the event")
}
delete(e.status, "/hooks")
e.tick(t)
before := len(e.hooksAt("/hooks"))
e.tick(t)
if len(e.hooksAt("/hooks")) != before {
t.Fatal("a delivered app endpoint was sent the event again")
}
if n := e.pendingOutbox(t); n != 0 {
t.Fatalf("%d events owed after delivery", n)
}
}
// Undelivered after 24 h: dropped, counted on the hook and shown to the
// owner, never sent; the event is no longer owed to the app.
func TestAppDelivery_DroppedAfter24Hours(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
e.item(t, e.companion.ID, "Old")
db := e.srv.store.DB()
// Owed (the hook was deliverable then), and older than 24 h.
if _, err := db.Exec(`UPDATE webhooks SET deliver_from = '1999-01-01T00:00:00Z' WHERE app_install_id = ?`, e.installID); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`UPDATE event_outbox SET occurred_at = '2000-01-01T00:00:00Z' WHERE dispatched_at IS NULL`); err != nil {
t.Fatal(err)
}
e.tick(t)
if got := e.hooksAt("/hooks"); len(got) != 0 {
t.Fatalf("a day-old event was delivered: %s", got[0].Body)
}
if n := e.pendingOutbox(t); n != 0 {
t.Fatalf("%d events owed after the drop", n)
}
got := e.getInstall(t)
if got.Webhook == nil || got.Webhook.UndeliveredDropped != 1 {
t.Fatalf("install view: %+v, want undelivered_dropped 1", got.Webhook)
}
// A second pass does not count it again.
e.tick(t)
if got := e.getInstall(t); got.Webhook.UndeliveredDropped != 1 {
t.Fatalf("undelivered_dropped %d after a second pass", got.Webhook.UndeliveredDropped)
}
}
func TestAppDelivery_DropBoundary(t *testing.T) {
now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
for _, tc := range []struct {
age time.Duration
want bool
}{{23 * time.Hour, false}, {24*time.Hour - time.Second, false}, {24*time.Hour + time.Second, true}, {25 * time.Hour, true}} {
if got := appDeliveryExpired(now.Add(-tc.age).Format(time.RFC3339), now); got != tc.want {
t.Errorf("age %v: expired %v, want %v", tc.age, got, tc.want)
}
}
if appDeliveryExpired("not a time", now) {
t.Error("an unparseable occurred_at was treated as expired")
}
}
// A skipped event (no projection) is recorded, so an owner hook's retry
// does not count it again.
func TestAppDelivery_SkipIsCountedOnce(t *testing.T) {
e := newDeliveryEnv(t)
m := metrics.New()
e.srv.SetMetrics(m)
e.redeemNow(t)
var down atomic.Int32
down.Store(http.StatusServiceUnavailable)
ownerSink(t, e, &down)
e.item(t, e.companion.ID, "No block")
db := e.srv.store.DB()
if _, err := db.Exec(`UPDATE event_outbox SET payload = '{"id":"x","title":"t"}' WHERE dispatched_at IS NULL`); err != nil {
t.Fatal(err)
}
e.tick(t)
e.tick(t)
var got dto.Metric
if err := m.AppWebhookDeliveriesTotal.WithLabelValues("skipped_no_projection").Write(&got); err != nil {
t.Fatal(err)
}
if n := got.GetCounter().GetValue(); n != 1 {
t.Fatalf("skipped_no_projection = %v across an owner retry, want 1", n)
}
}
func (e *deliveryEnv) setAppRate(t *testing.T, limit rate.Limit, burst int) {
t.Helper()
e.srv.appRate.mu.Lock()
e.srv.appRate.by, e.srv.appRate.limit, e.srv.appRate.burst = map[string]*rate.Limiter{}, limit, burst
e.srv.appRate.mu.Unlock()
}
// codex r1 on U10c: an event in a collection the app is not subscribed to
// spends none of its rate budget.
func TestAppDelivery_UnrelatedCollectionsSpendNoBudget(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
e.setAppRate(t, 0, 1) // one token, never refilled
other, err := e.srv.store.CreateCollection(e.wsID, models.CollectionCreate{Name: "Internal", Slug: "internal"})
if err != nil {
t.Fatal(err)
}
e.item(t, other.ID, "Not the app's")
e.tick(t)
e.item(t, e.companion.ID, "The app's")
e.tick(t)
if got := e.hooksAt("/hooks"); len(got) != 1 {
t.Fatalf("%d deliveries: an unrelated event spent the app's only token", len(got))
}
}
// codex r1 on U10c: the cap bounds REQUESTS, retries included.
func TestAppDelivery_RateCapCountsEveryAttempt(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
e.setAppRate(t, 0, 2) // two requests, never refilled
e.status["/hooks"] = http.StatusServiceUnavailable
e.item(t, e.companion.ID, "Flaky")
e.tick(t)
if got := len(e.hooksAt("/hooks")); got != 2 {
t.Fatalf("%d requests against a budget of 2", got)
}
if e.pendingOutbox(t) == 0 {
t.Fatal("the event was acked while still owed")
}
}
// codex r1 on U10c: the limiter map does not grow without bound, and pruning
// drops only idle (full-bucket) entries.
func TestAppRateLimiter_PrunesIdle(t *testing.T) {
l := appRateLimiter{by: map[string]*rate.Limiter{}, limit: 10, burst: 60}
busy := rate.NewLimiter(l.limit, l.burst)
busy.AllowN(time.Now(), 60) // spent: not idle
l.by["busy"] = busy
for i := 0; len(l.by) < appRateLimiterPruneAt; i++ {
l.by[fmt.Sprintf("idle-%d", i)] = rate.NewLimiter(l.limit, l.burst) // full: idle
}
l.allow("new")
if len(l.by) > 2 {
t.Fatalf("%d limiters retained after pruning, want busy and new", len(l.by))
}
if _, ok := l.by["busy"]; !ok {
t.Fatal("pruning dropped a limiter that had spent tokens")
}
}
// codex r1 on U10c: a bulk skip is recorded, so an owner hook's retry of
// the same unit does not count it again.
func TestAppDelivery_BulkSkipCountedOnce(t *testing.T) {
e := newDeliveryEnv(t)
m := metrics.New()
e.srv.SetMetrics(m)
e.redeemNow(t)
var down atomic.Int32
down.Store(http.StatusServiceUnavailable)
ownerSink(t, e, &down)
it := e.item(t, e.companion.ID, "Member")
e.tick(t)
title := "Renamed in bulk"
if _, err := e.srv.store.UpdateItem(it.ID, models.ItemUpdate{Title: &title}, store.WithEventBatch("batch-once")); err != nil {
t.Fatal(err)
}
e.tick(t)
e.tick(t)
var got dto.Metric
if err := m.AppWebhookDeliveriesTotal.WithLabelValues("skipped_bulk").Write(&got); err != nil {
t.Fatal(err)
}
if n := got.GetCounter().GetValue(); n != 1 {
t.Fatalf("skipped_bulk = %v across an owner retry, want 1", n)
}
}
// codex r2 on U10c: an old event the app was never owed (before its
// deliver_from) is refused, not dropped and counted.
func TestAppDelivery_NotOwedIsNotDropped(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
e.item(t, e.companion.ID, "Before the hook")
if _, err := e.srv.store.DB().Exec(`UPDATE event_outbox SET occurred_at = '2000-01-01T00:00:00Z' WHERE dispatched_at IS NULL`); err != nil {
t.Fatal(err)
}
e.tick(t)
if got := e.getInstall(t); got.Webhook.UndeliveredDropped != 0 {
t.Fatalf("undelivered_dropped %d for an event the app was never owed", got.Webhook.UndeliveredDropped)
}
if n := e.pendingOutbox(t); n != 0 {
t.Fatalf("%d events owed", n)
}
}
// codex r2 on U10c: an attempt admission refuses spends no rate budget.
func TestAppDelivery_RefusalsSpendNoBudget(t *testing.T) {
e := newDeliveryEnv(t)
var down atomic.Int32
down.Store(http.StatusServiceUnavailable)
ownerSink(t, e, &down)
e.item(t, e.companion.ID, "While held") // pending: the owner is down
e.tick(t)
time.Sleep(1100 * time.Millisecond)
e.redeemNow(t)
e.setAppRate(t, 0, 1) // one token
e.item(t, e.companion.ID, "After redeem")
e.tick(t)
if got := e.hooksAt("/hooks"); len(got) != 1 {
t.Fatalf("%d deliveries: the refused old event spent the only token", len(got))
}
}
// codex r2 on U10c: the ledger counts requests actually sent.
func TestAppDelivery_LedgerCountsRequests(t *testing.T) {
e := newDeliveryEnv(t)
e.redeemNow(t)
e.status["/hooks"] = http.StatusServiceUnavailable
e.item(t, e.companion.ID, "Three tries")
e.tick(t)
attempts := func() (string, int) {
var st string
var n int
if err := e.srv.store.DB().QueryRow(`SELECT d.status, d.attempts FROM webhook_deliveries d JOIN webhooks w ON w.id = d.webhook_id WHERE w.app_install_id = ?`, e.installID).Scan(&st, &n); err != nil {
t.Fatal(err)
}
return st, n
}
if st, n := attempts(); st != "transient" || n != 3 {
t.Fatalf("ledger %s/%d after three 503s, want transient/3", st, n)
}
// One more request, then the cap.
e.setAppRate(t, 0, 1)
e.tick(t)
if st, n := attempts(); st != "rate_limited" || n != 4 {
t.Fatalf("ledger %s/%d, want rate_limited/4 (the request sent before the cap counts)", st, n)
}
}
+2
View File
@@ -198,6 +198,8 @@ type Server struct {
appFetchTLS *tls.Config
// appPosters caches the app webhook Poster (TASK-3408 U10b).
appPosters appPosterCache
// appRate is the per-install app webhook rate cap (TASK-3408 U10c).
appRate appRateLimiter
// webhookCancel ends the dispatcher's context at Stop, so backoff waits
// and app attempts do not outlive the server (TASK-3408 U10b).
webhookCancel context.CancelFunc
+149 -39
View File
@@ -45,6 +45,16 @@ var appDeliveryDrainPoll = 250 * time.Millisecond
type AppHookTarget struct {
WebhookID string
InstallID string
// Events is the stored subscription list, so the caller can narrow to
// the event's collection before spending anything on a target
// (AppHookSubscribes; codex r1 on U10c).
Events string
}
// AppHookSubscribes reports whether a stored subscription list names event
// for collectionID (any collection when collectionID is "").
func AppHookSubscribes(eventsJSON, event, collectionID string) bool {
return appHookSubscribes(eventsJSON, event, collectionID)
}
// ListAppWebhookTargets returns the workspace's deliverable-looking app hooks
@@ -63,11 +73,10 @@ func (s *Store) ListAppWebhookTargets(workspaceID, event string) ([]AppHookTarge
var out []AppHookTarget
for rows.Next() {
var t AppHookTarget
var events string
if err := rows.Scan(&t.WebhookID, &t.InstallID, &events); err != nil {
if err := rows.Scan(&t.WebhookID, &t.InstallID, &t.Events); err != nil {
return nil, err
}
if appHookSubscribes(events, event, "") {
if appHookSubscribes(t.Events, event, "") {
out = append(out, t)
}
}
@@ -128,13 +137,9 @@ func (e *AppDeliveryRefusedError) Error() string { return "app delivery refused:
func (s *Store) AdmitAppDelivery(webhookID, event, collectionID, occurredAt, deliveryID string) (*AppDeliveryAdmission, error) {
// The hook's install never changes (app_install_id is written once), so
// it is read before the lock to know which install row to take.
var installID sql.NullString
err := s.db.QueryRow(s.q(`SELECT app_install_id FROM webhooks WHERE id = ?`), webhookID).Scan(&installID)
if errors.Is(err, sql.ErrNoRows) || (err == nil && !installID.Valid) {
return nil, &AppDeliveryRefusedError{Reason: "hook_gone"}
}
installID, err := s.appHookInstall(webhookID)
if err != nil {
return nil, fmt.Errorf("admit app delivery: %w", err)
return nil, err
}
tx, err := s.db.Begin()
@@ -143,12 +148,136 @@ func (s *Store) AdmitAppDelivery(webhookID, event, collectionID, occurredAt, del
}
defer tx.Rollback()
q := `SELECT state, workspace_id FROM app_installs WHERE id = ?`
hook, err := s.appDeliveryChecksQ(tx, installID, webhookID, event, collectionID, occurredAt, true)
if err != nil {
return nil, err
}
insert := `INSERT INTO app_delivery_inflight (delivery_id, install_id, expires_at)
VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+12 seconds'))`
if s.dialect.Driver() == DriverPostgres {
q += ` FOR SHARE`
insert = `INSERT INTO app_delivery_inflight (delivery_id, install_id, expires_at)
VALUES (?, ?, now() + interval '12 seconds')`
}
if _, err := tx.Exec(s.q(insert), deliveryID, installID); err != nil {
return nil, fmt.Errorf("admit app delivery: in-flight row: %w", err)
}
if err := tx.Commit(); err != nil {
return nil, fmt.Errorf("admit app delivery: commit: %w", err)
}
plain, err := s.decrypt(hook.secret)
if err != nil {
// Admitted but unsendable: release the row now rather than make the
// drain wait for it.
_ = s.EndAppDelivery(deliveryID)
return nil, fmt.Errorf("admit app delivery: decrypt secret: %w", err)
}
return &AppDeliveryAdmission{URL: hook.url, Secret: plain, InstallID: installID}, nil
}
// DropOwedDelivery decides a delivery that is past the drop age (codex r2/r4
// on U10c). In ONE transaction that takes the install row first, as
// admission does, it runs admission's checks: a delivery admission would
// refuse was never owed and is recorded refused; one it would admit is
// recorded dropped and counted on the hook. A terminal row is left alone.
// It returns "dropped", "refused:<reason>" or "" (already terminal).
func (s *Store) DropOwedDelivery(eventID, webhookID, event, collectionID, occurredAt, reason string) (string, error) {
installID, err := s.appHookInstall(webhookID)
var refused *AppDeliveryRefusedError
if errors.As(err, &refused) {
return s.recordDropDecision(eventID, webhookID, "refused:"+refused.Reason)
}
if err != nil {
return "", err
}
tx, err := s.db.Begin()
if err != nil {
return "", err
}
defer tx.Rollback()
outcome := "dropped"
if _, err := s.appDeliveryChecksQ(tx, installID, webhookID, event, collectionID, occurredAt, true); err != nil {
if !errors.As(err, &refused) {
return "", err
}
outcome = "refused:" + refused.Reason
}
done, err := s.recordDropDecisionTx(tx, eventID, webhookID, outcome, reason)
if err != nil {
return "", err
}
if err := tx.Commit(); err != nil {
return "", err
}
return done, nil
}
func (s *Store) recordDropDecision(eventID, webhookID, outcome string) (string, error) {
tx, err := s.db.Begin()
if err != nil {
return "", err
}
defer tx.Rollback()
done, err := s.recordDropDecisionTx(tx, eventID, webhookID, outcome, "")
if err != nil {
return "", err
}
return done, tx.Commit()
}
// recordDropDecisionTx writes the terminal row for a drop-age decision and,
// for a drop, counts it on the hook. "" when the row was already terminal.
func (s *Store) recordDropDecisionTx(tx *sql.Tx, eventID, webhookID, outcome, reason string) (string, error) {
if dropDecisionHook != nil {
dropDecisionHook()
}
status, lastError := DeliveryDropped, reason
if outcome != "dropped" {
status, lastError = DeliveryRefused, outcome
}
res, err := tx.Exec(s.q(`INSERT INTO webhook_deliveries (outbox_event_id, webhook_id, status, attempts, last_error, updated_at)
VALUES (?, ?, ?, 0, ?, ?)
ON CONFLICT (outbox_event_id, webhook_id) DO UPDATE SET status = excluded.status, last_error = excluded.last_error, updated_at = excluded.updated_at
WHERE webhook_deliveries.status NOT IN ('delivered', 'permanent', 'refused', 'skipped', 'dropped')`),
eventID, webhookID, status, nullIfEmpty(lastError), now())
if err != nil {
return "", fmt.Errorf("drop decision: %w", err)
}
if n, _ := res.RowsAffected(); n == 0 {
return "", nil
}
if status == DeliveryDropped {
if _, err := tx.Exec(s.q(`UPDATE webhooks SET dropped_count = dropped_count + 1 WHERE id = ?`), webhookID); err != nil {
return "", fmt.Errorf("drop decision: count: %w", err)
}
}
return outcome, nil
}
func (s *Store) appHookInstall(webhookID string) (string, error) {
var installID sql.NullString
err := s.db.QueryRow(s.q(`SELECT app_install_id FROM webhooks WHERE id = ?`), webhookID).Scan(&installID)
if errors.Is(err, sql.ErrNoRows) || (err == nil && !installID.Valid) {
return "", &AppDeliveryRefusedError{Reason: "hook_gone"}
}
if err != nil {
return "", fmt.Errorf("admit app delivery: %w", err)
}
return installID.String, nil
}
type appHookRow struct{ url, secret string }
// appDeliveryChecksQ is admission's decision, on q. lock takes the install
// row FOR SHARE first (Postgres; admission runs it in a transaction); the
// read-only pre-check passes false. One function so the two cannot drift.
func (s *Store) appDeliveryChecksQ(q Queryer, installID, webhookID, event, collectionID, occurredAt string, lock bool) (*appHookRow, error) {
iq := `SELECT state, workspace_id FROM app_installs WHERE id = ?`
if lock && s.dialect.Driver() == DriverPostgres {
iq += ` FOR SHARE`
}
var state, workspaceID string
err = tx.QueryRow(s.q(q), installID.String).Scan(&state, &workspaceID)
err := q.QueryRow(s.q(iq), installID).Scan(&state, &workspaceID)
if errors.Is(err, sql.ErrNoRows) {
return nil, &AppDeliveryRefusedError{Reason: "install_gone"}
}
@@ -163,7 +292,7 @@ func (s *Store) AdmitAppDelivery(webhookID, event, collectionID, occurredAt, del
// WorkspaceLive check, BUG-3340). A plain read: deletion that races this
// serializes as "delivered, then deleted", the BUG-3340 residual.
var live int
if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM workspaces WHERE id = ? AND deleted_at IS NULL`), workspaceID).Scan(&live); err != nil {
if err := q.QueryRow(s.q(`SELECT COUNT(*) FROM workspaces WHERE id = ? AND deleted_at IS NULL`), workspaceID).Scan(&live); err != nil {
return nil, fmt.Errorf("admit app delivery: workspace: %w", err)
}
if live == 0 {
@@ -173,10 +302,11 @@ func (s *Store) AdmitAppDelivery(webhookID, event, collectionID, occurredAt, del
// Under the install lock: an upgrade (which changes the URL and the
// subscriptions) and uninstall (which deletes the hook) both hold this
// row FOR UPDATE, so what is read here is what the owner last consented.
var url, secret, events string
var row appHookRow
var events string
var delivered, deliverFrom sql.NullString
err = tx.QueryRow(s.q(`SELECT url, secret, events, secret_delivered_at, deliver_from FROM webhooks WHERE id = ?`), webhookID).
Scan(&url, &secret, &events, &delivered, &deliverFrom)
err = q.QueryRow(s.q(`SELECT url, secret, events, secret_delivered_at, deliver_from FROM webhooks WHERE id = ?`), webhookID).
Scan(&row.url, &row.secret, &events, &delivered, &deliverFrom)
if errors.Is(err, sql.ErrNoRows) {
return nil, &AppDeliveryRefusedError{Reason: "hook_gone"}
}
@@ -195,34 +325,14 @@ func (s *Store) AdmitAppDelivery(webhookID, event, collectionID, occurredAt, del
// The ceiling: the event's collection must still be this install's
// companion. A released or deleted companion is no longer the app's.
var n int
if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM collections WHERE id = ? AND via_app = ? AND deleted_at IS NULL`),
collectionID, installID.String).Scan(&n); err != nil {
if err := q.QueryRow(s.q(`SELECT COUNT(*) FROM collections WHERE id = ? AND via_app = ? AND deleted_at IS NULL`),
collectionID, installID).Scan(&n); err != nil {
return nil, fmt.Errorf("admit app delivery: ceiling: %w", err)
}
if n == 0 {
return nil, &AppDeliveryRefusedError{Reason: "not_visible"}
}
insert := `INSERT INTO app_delivery_inflight (delivery_id, install_id, expires_at)
VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+12 seconds'))`
if s.dialect.Driver() == DriverPostgres {
insert = `INSERT INTO app_delivery_inflight (delivery_id, install_id, expires_at)
VALUES (?, ?, now() + interval '12 seconds')`
}
if _, err := tx.Exec(s.q(insert), deliveryID, installID.String); err != nil {
return nil, fmt.Errorf("admit app delivery: in-flight row: %w", err)
}
if err := tx.Commit(); err != nil {
return nil, fmt.Errorf("admit app delivery: commit: %w", err)
}
plain, err := s.decrypt(secret)
if err != nil {
// Admitted but unsendable: release the row now rather than make the
// drain wait for it.
_ = s.EndAppDelivery(deliveryID)
return nil, fmt.Errorf("admit app delivery: decrypt secret: %w", err)
}
return &AppDeliveryAdmission{URL: url, Secret: plain, InstallID: installID.String}, nil
return &row, nil
}
// EndAppDelivery removes an attempt's in-flight row. Safe to repeat.
+6 -2
View File
@@ -44,6 +44,9 @@ type AppWebhookStatus struct {
// Status is "awaiting_secret" (held: no redeem has handed the app its
// signing secret yet) or "active" (deliverable while the install is).
Status string `json:"status"`
// Dropped counts deliveries given up undelivered after
// appWebhookDropAfter (TASK-3408 U10c).
Dropped int64 `json:"undelivered_dropped"`
}
const (
@@ -176,14 +179,15 @@ func (s *Store) holdAppWebhookTx(tx *sql.Tx, installID string) error {
func (s *Store) GetAppWebhookStatus(installID string) (*AppWebhookStatus, error) {
var url string
var delivered sql.NullString
err := s.db.QueryRow(s.q(`SELECT url, secret_delivered_at FROM webhooks WHERE app_install_id = ?`), installID).Scan(&url, &delivered)
var dropped int64
err := s.db.QueryRow(s.q(`SELECT url, secret_delivered_at, dropped_count FROM webhooks WHERE app_install_id = ?`), installID).Scan(&url, &delivered, &dropped)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("app webhook status: %w", err)
}
st := &AppWebhookStatus{URL: url, Status: AppWebhookAwaitingSecret}
st := &AppWebhookStatus{URL: url, Status: AppWebhookAwaitingSecret, Dropped: dropped}
if delivered.Valid && delivered.String != "" {
st.Status = AppWebhookActive
}
@@ -0,0 +1,27 @@
-- Migration 124: per-endpoint delivery state for app webhooks (SPEC-6 U10c,
-- DOC-3371 §5; TASK-3408).
--
-- One row per (outbox event, app hook) the drain has decided anything about.
-- A TERMINAL row (delivered, permanent, refused, skipped, dropped) is never
-- retried, so a retry goes only to endpoints still owed, and an event the
-- app was refused (held, disabled, not visible) stays refused. A PENDING row
-- (transient, deferred, rate_limited) keeps the event owed until it turns
-- terminal or is dropped after appWebhookDropAfter.
--
-- Both foreign keys cascade: the delivery state belongs to its outbox row
-- (retention deletes them together) and to its hook (uninstall deletes it).
-- The outbox's own no-foreign-key rule is about its SUBJECTS, not this.
CREATE TABLE IF NOT EXISTS webhook_deliveries (
outbox_event_id TEXT NOT NULL REFERENCES event_outbox(id) ON DELETE CASCADE,
webhook_id TEXT NOT NULL REFERENCES webhooks(id) ON DELETE CASCADE,
status TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
updated_at TEXT NOT NULL,
PRIMARY KEY (outbox_event_id, webhook_id)
);
CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_hook ON webhook_deliveries (webhook_id, status);
-- Deliveries dropped undelivered after appWebhookDropAfter, shown to the
-- owner on the install.
ALTER TABLE webhooks ADD COLUMN dropped_count INTEGER NOT NULL DEFAULT 0;
@@ -352,6 +352,11 @@ watches.id
watches.item_id
watches.user_id
watches.workspace_id
webhook_deliveries.last_error
webhook_deliveries.outbox_event_id
webhook_deliveries.status
webhook_deliveries.updated_at
webhook_deliveries.webhook_id
webhooks.app_install_id
webhooks.created_at
webhooks.deliver_from
@@ -0,0 +1,13 @@
-- Migration 098: per-endpoint delivery state for app webhooks (SPEC-6 U10c,
-- TASK-3408). Mirrors SQLite migration 124; see it for the rationale.
CREATE TABLE IF NOT EXISTS webhook_deliveries (
outbox_event_id TEXT NOT NULL REFERENCES event_outbox(id) ON DELETE CASCADE,
webhook_id TEXT NOT NULL REFERENCES webhooks(id) ON DELETE CASCADE,
status TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
updated_at TEXT NOT NULL,
PRIMARY KEY (outbox_event_id, webhook_id)
);
CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_hook ON webhook_deliveries (webhook_id, status);
ALTER TABLE webhooks ADD COLUMN IF NOT EXISTS dropped_count INTEGER NOT NULL DEFAULT 0;
+272
View File
@@ -0,0 +1,272 @@
package store
import (
"fmt"
"sync"
"testing"
"time"
"github.com/PerpetualSoftware/pad/internal/models"
)
// TASK-3408 (U10c): the delivery ledger. The first terminal decision
// stands; attempts count only real attempts; a drop is counted once.
func TestTask3408_DeliveryLedger(t *testing.T) {
f := task3408Fixture(t, "inst-ledger")
item, err := f.s.CreateItem(f.ws.ID, f.companion.ID, models.ItemCreate{Title: "T", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
var eventID string
if err := f.s.db.QueryRow(f.s.q(`SELECT id FROM event_outbox WHERE subject_id = ? ORDER BY occurred_at DESC LIMIT 1`), item.ID).Scan(&eventID); err != nil {
t.Fatal(err)
}
row := func() (string, int) {
var st string
var n int
if err := f.s.db.QueryRow(f.s.q(`SELECT status, attempts FROM webhook_deliveries WHERE outbox_event_id = ? AND webhook_id = ?`), eventID, f.hookID).Scan(&st, &n); err != nil {
t.Fatal(err)
}
return st, n
}
must := func(err error) {
t.Helper()
if err != nil {
t.Fatal(err)
}
}
must(f.s.RecordDelivery(eventID, f.hookID, DeliveryRateLimited, "cap", 0))
must(f.s.RecordDelivery(eventID, f.hookID, DeliveryTransient, "503", 1))
if st, n := row(); st != DeliveryTransient || n != 1 {
t.Fatalf("pending row: %s/%d, want transient/1 (a deferral is not an attempt)", st, n)
}
must(f.s.RecordDelivery(eventID, f.hookID, DeliveryDelivered, "", 1))
must(f.s.RecordDelivery(eventID, f.hookID, DeliveryTransient, "late", 1))
if st, n := row(); st != DeliveryDelivered || n != 2 {
t.Fatalf("after a late write: %s/%d, want delivered/2 (terminal stands)", st, n)
}
if outcome, err := f.s.DropOwedDelivery(eventID, f.hookID, "item.created", f.companion.ID, now(), "old"); err != nil || outcome != "" {
t.Fatalf("dropping a delivered row: %q %v", outcome, err)
}
var count int
must(f.s.db.QueryRow(f.s.q(`SELECT dropped_count FROM webhooks WHERE id = ?`), f.hookID).Scan(&count))
if count != 0 {
t.Fatalf("dropped_count %d after a refused drop", count)
}
// A pending row drops once, and counts once.
if _, err := f.s.db.Exec(f.s.q(`DELETE FROM webhook_deliveries WHERE outbox_event_id = ?`), eventID); err != nil {
t.Fatal(err)
}
must(f.s.RecordDelivery(eventID, f.hookID, DeliveryTransient, "503", 1))
for i := 0; i < 2; i++ {
if _, err := f.s.DropOwedDelivery(eventID, f.hookID, "item.created", f.companion.ID, now(), "old"); err != nil {
t.Fatal(err)
}
}
must(f.s.db.QueryRow(f.s.q(`SELECT dropped_count FROM webhooks WHERE id = ?`), f.hookID).Scan(&count))
if st, _ := row(); st != DeliveryDropped || count != 1 {
t.Fatalf("drop: %s, dropped_count %d, want dropped/1", st, count)
}
// Owner rows: only delivered is recorded, and it is read back.
if ok, err := f.s.OwnerDelivered(eventID, "no-hook"); err != nil || ok {
t.Fatalf("an unrecorded owner hook: %v %v", ok, err)
}
}
// codex r4 on U10c: the drop decision runs admission's checks under the
// install lock, so a delivery no longer owed (here: the install disabled
// since) is recorded refused, not dropped and counted.
func TestTask3408_DropOfAnUnowedDeliveryIsARefusal(t *testing.T) {
f := task3408Fixture(t, "inst-dropref")
item, err := f.s.CreateItem(f.ws.ID, f.companion.ID, models.ItemCreate{Title: "T", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
var eventID string
if err := f.s.db.QueryRow(f.s.q(`SELECT id FROM event_outbox WHERE subject_id = ? ORDER BY occurred_at DESC LIMIT 1`), item.ID).Scan(&eventID); err != nil {
t.Fatal(err)
}
if err := f.s.BeginInstallTeardown(f.ws.ID, f.installID, TeardownDisable); err != nil {
t.Fatal(err)
}
outcome, err := f.s.DropOwedDelivery(eventID, f.hookID, "item.created", f.companion.ID, now(), "old")
if err != nil || outcome != "refused:install_disabling" {
t.Fatalf("outcome %q (%v), want refused:install_disabling", outcome, err)
}
var count int
if err := f.s.db.QueryRow(f.s.q(`SELECT dropped_count FROM webhooks WHERE id = ?`), f.hookID).Scan(&count); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatalf("dropped_count %d for a delivery that was not owed", count)
}
}
// codex r5 on U10c: a request admitted before another drainer dropped the
// event did reach the app; "delivered" replaces "dropped" and the drop is
// taken back off the count. Any other terminal row still stands.
func TestTask3408_DeliveredOverridesADrop(t *testing.T) {
f := task3408Fixture(t, "inst-undrop")
item, err := f.s.CreateItem(f.ws.ID, f.companion.ID, models.ItemCreate{Title: "T", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
var eventID string
if err := f.s.db.QueryRow(f.s.q(`SELECT id FROM event_outbox WHERE subject_id = ? ORDER BY occurred_at DESC LIMIT 1`), item.ID).Scan(&eventID); err != nil {
t.Fatal(err)
}
if out, err := f.s.DropOwedDelivery(eventID, f.hookID, "item.created", f.companion.ID, now(), "old"); err != nil || out != "dropped" {
t.Fatalf("drop: %q %v", out, err)
}
if err := f.s.RecordDelivery(eventID, f.hookID, DeliveryDelivered, "", 1); err != nil {
t.Fatal(err)
}
var st string
var count int
if err := f.s.db.QueryRow(f.s.q(`SELECT d.status, w.dropped_count FROM webhook_deliveries d JOIN webhooks w ON w.id = d.webhook_id WHERE d.outbox_event_id = ? AND d.webhook_id = ?`), eventID, f.hookID).Scan(&st, &count); err != nil {
t.Fatal(err)
}
if st != DeliveryDelivered || count != 0 {
t.Fatalf("after a late delivery: %s, dropped_count %d; want delivered, 0", st, count)
}
// Only delivered may replace dropped.
if err := f.s.RecordDelivery(eventID, f.hookID, DeliveryPermanent, "late", 1); err != nil {
t.Fatal(err)
}
if err := f.s.db.QueryRow(f.s.q(`SELECT status FROM webhook_deliveries WHERE outbox_event_id = ? AND webhook_id = ?`), eventID, f.hookID).Scan(&st); err != nil {
t.Fatal(err)
}
if st != DeliveryDelivered {
t.Fatalf("a terminal delivered row was overwritten with %s", st)
}
// A drop followed by anything but a delivery stays dropped.
item2, err := f.s.CreateItem(f.ws.ID, f.companion.ID, models.ItemCreate{Title: "T2", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
var event2 string
if err := f.s.db.QueryRow(f.s.q(`SELECT id FROM event_outbox WHERE subject_id = ? ORDER BY occurred_at DESC LIMIT 1`), item2.ID).Scan(&event2); err != nil {
t.Fatal(err)
}
if out, err := f.s.DropOwedDelivery(event2, f.hookID, "item.created", f.companion.ID, now(), "old"); err != nil || out != "dropped" {
t.Fatalf("drop: %q %v", out, err)
}
for _, late := range []string{DeliveryPermanent, DeliveryTransient} {
if err := f.s.RecordDelivery(event2, f.hookID, late, "late", 1); err != nil {
t.Fatal(err)
}
}
if err := f.s.db.QueryRow(f.s.q(`SELECT status FROM webhook_deliveries WHERE outbox_event_id = ? AND webhook_id = ?`), event2, f.hookID).Scan(&st); err != nil {
t.Fatal(err)
}
if st != DeliveryDropped {
t.Fatalf("a dropped row was overwritten with %s", st)
}
}
// codex r6 on U10c: a drop and a late delivery racing on a row that does
// not exist yet. Whatever the interleaving, dropped_count equals the number
// of events whose final row is dropped (Postgres is where this bites; SQLite
// serializes the writers).
func TestTask3408_DropAndDeliveryRace(t *testing.T) {
f := task3408Fixture(t, "inst-race2")
const rounds = 30
events := make([]string, 0, rounds)
for i := 0; i < rounds; i++ {
item, err := f.s.CreateItem(f.ws.ID, f.companion.ID, models.ItemCreate{Title: fmt.Sprintf("R%d", i), Fields: `{}`})
if err != nil {
t.Fatal(err)
}
var id string
if err := f.s.db.QueryRow(f.s.q(`SELECT id FROM event_outbox WHERE subject_id = ? ORDER BY occurred_at DESC LIMIT 1`), item.ID).Scan(&id); err != nil {
t.Fatal(err)
}
events = append(events, id)
}
for _, ev := range events {
var wg sync.WaitGroup
errs := make(chan error, 2)
wg.Add(2)
go func() {
defer wg.Done()
_, err := f.s.DropOwedDelivery(ev, f.hookID, "item.created", f.companion.ID, now(), "old")
errs <- err
}()
go func() {
defer wg.Done()
errs <- f.s.RecordDelivery(ev, f.hookID, DeliveryDelivered, "", 1)
}()
wg.Wait()
close(errs)
for err := range errs {
if err != nil {
t.Fatal(err)
}
}
}
var dropped, count int
if err := f.s.db.QueryRow(f.s.q(`SELECT COUNT(*) FROM webhook_deliveries WHERE webhook_id = ? AND status = 'dropped'`), f.hookID).Scan(&dropped); err != nil {
t.Fatal(err)
}
if err := f.s.db.QueryRow(f.s.q(`SELECT dropped_count FROM webhooks WHERE id = ?`), f.hookID).Scan(&count); err != nil {
t.Fatal(err)
}
if count != dropped {
t.Fatalf("dropped_count %d, but %d rows are dropped", count, dropped)
}
}
// The same race made deterministic: a drop starts while RecordDelivery is
// between its first statement and its commit. The drop must wait for the
// delivery and then leave it alone; dropped_count stays 0.
func TestTask3408_DropDuringDeliveryRecord(t *testing.T) {
f := task3408Fixture(t, "inst-race3")
item, err := f.s.CreateItem(f.ws.ID, f.companion.ID, models.ItemCreate{Title: "R", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
var ev string
if err := f.s.db.QueryRow(f.s.q(`SELECT id FROM event_outbox WHERE subject_id = ? ORDER BY occurred_at DESC LIMIT 1`), item.ID).Scan(&ev); err != nil {
t.Fatal(err)
}
dropDone := make(chan error, 1)
reached := make(chan struct{}, 1)
dropDecisionHook = func() { reached <- struct{}{} }
recordDeliveryHook = func() {
go func() {
_, err := f.s.DropOwedDelivery(ev, f.hookID, "item.created", f.companion.ID, now(), "old")
dropDone <- err
}()
if f.s.dialect.Driver() == DriverPostgres {
// The drop holds the install lock and is about to write its row:
// it must now block on the row this transaction inserted.
select {
case <-reached:
case <-time.After(10 * time.Second):
t.Error("the drop never reached its write")
}
time.Sleep(100 * time.Millisecond) // let its upsert be issued
}
// SQLite: the drop cannot begin while this transaction holds the
// write lock (BEGIN IMMEDIATE), so it runs after the commit.
}
t.Cleanup(func() { recordDeliveryHook, dropDecisionHook = nil, nil })
if err := f.s.RecordDelivery(ev, f.hookID, DeliveryDelivered, "", 1); err != nil {
t.Fatal(err)
}
recordDeliveryHook = nil
if err := <-dropDone; err != nil {
t.Fatal(err)
}
var st string
var count int
if err := f.s.db.QueryRow(f.s.q(`SELECT d.status, w.dropped_count FROM webhook_deliveries d JOIN webhooks w ON w.id = d.webhook_id WHERE d.outbox_event_id = ? AND d.webhook_id = ?`), ev, f.hookID).Scan(&st, &count); err != nil {
t.Fatal(err)
}
if st != DeliveryDelivered || count != 0 {
t.Fatalf("final %s, dropped_count %d; want delivered, 0", st, count)
}
}
+134
View File
@@ -0,0 +1,134 @@
package store
import (
"database/sql"
"errors"
"fmt"
)
// Per-endpoint webhook delivery state (SPEC-6 U10c, DOC-3371 §5; TASK-3408).
//
// For APP hooks every decision is recorded, and a terminal row is never
// retried. For OWNER hooks only "delivered" is recorded, and used only to
// skip re-sending an event an owner endpoint already received when the
// event stays owed to someone else (an app's rate limit or outage must not
// duplicate deliveries to people who never installed it; lead ruling, day
// 87). Owner failure, retry and ack are otherwise unchanged; full owner
// per-endpoint state is TASK-3409.
// Delivery statuses. Terminal ones end the endpoint's part in the event.
const (
DeliveryDelivered = "delivered"
DeliveryPermanent = "permanent"
DeliveryRefused = "refused"
DeliverySkipped = "skipped"
DeliveryDropped = "dropped"
DeliveryTransient = "transient"
DeliveryDeferred = "deferred"
DeliveryRateLimited = "rate_limited"
)
// DeliveryTerminal reports whether a status ends the endpoint's part.
func DeliveryTerminal(status string) bool {
switch status {
case DeliveryDelivered, DeliveryPermanent, DeliveryRefused, DeliverySkipped, DeliveryDropped:
return true
}
return false
}
// DeliveryStatus returns the recorded status of eventID for webhookID, ""
// when nothing is recorded.
func (s *Store) DeliveryStatus(eventID, webhookID string) (string, error) {
var st string
err := s.db.QueryRow(s.q(`SELECT status FROM webhook_deliveries WHERE outbox_event_id = ? AND webhook_id = ?`), eventID, webhookID).Scan(&st)
if errors.Is(err, sql.ErrNoRows) {
return "", nil
}
if err != nil {
return "", fmt.Errorf("delivery status: %w", err)
}
return st, nil
}
// RecordDelivery stores status for (eventID, webhookID), adding attempts
// requests actually sent (a refusal or a rate-limited deferral sends none).
// A terminal row is never overwritten, the first terminal decision standing,
// with ONE exception: "delivered" replaces "dropped" and takes the drop back
// off the hook's count. A request admitted before another drainer dropped
// the event (its claim expired mid-request) did reach the app, and the
// owner must not see it as lost (codex r5 on U10c).
func (s *Store) RecordDelivery(eventID, webhookID, status, lastError string, attempts int) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
ts := now()
// Insert first: ON CONFLICT waits for a concurrent inserter (a drop) to
// commit, so a row that exists afterwards can be locked and read, and an
// absent one is simply ours (codex r6 on U10c: FOR UPDATE alone locks
// nothing when the row does not exist yet).
res, err := tx.Exec(s.q(`INSERT INTO webhook_deliveries (outbox_event_id, webhook_id, status, attempts, last_error, updated_at)
VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (outbox_event_id, webhook_id) DO NOTHING`),
eventID, webhookID, status, attempts, nullIfEmpty(lastError), ts)
if err != nil {
return fmt.Errorf("record delivery: insert: %w", err)
}
if recordDeliveryHook != nil {
recordDeliveryHook()
}
if n, _ := res.RowsAffected(); n == 1 {
return tx.Commit()
}
q := `SELECT status FROM webhook_deliveries WHERE outbox_event_id = ? AND webhook_id = ?`
if s.dialect.Driver() == DriverPostgres {
q += ` FOR UPDATE`
}
var prev string
if err := tx.QueryRow(s.q(q), eventID, webhookID).Scan(&prev); err != nil {
return fmt.Errorf("record delivery: read: %w", err)
}
undrop := prev == DeliveryDropped && status == DeliveryDelivered
if DeliveryTerminal(prev) && !undrop {
return nil
}
if _, err := tx.Exec(s.q(`UPDATE webhook_deliveries SET status = ?, attempts = attempts + ?, last_error = ?, updated_at = ?
WHERE outbox_event_id = ? AND webhook_id = ?`),
status, attempts, nullIfEmpty(lastError), ts, eventID, webhookID); err != nil {
return fmt.Errorf("record delivery: %w", err)
}
if undrop {
if _, err := tx.Exec(s.q(`UPDATE webhooks SET dropped_count = dropped_count - 1 WHERE id = ? AND dropped_count > 0`), webhookID); err != nil {
return fmt.Errorf("record delivery: undo drop: %w", err)
}
}
return tx.Commit()
}
// OwnerDelivered reports whether an owner hook already received eventID.
func (s *Store) OwnerDelivered(eventID, webhookID string) (bool, error) {
st, err := s.DeliveryStatus(eventID, webhookID)
return st == DeliveryDelivered, err
}
// RecordOwnerDelivered records that an owner hook received eventID. Only
// success is recorded for owner hooks (see the file comment).
func (s *Store) RecordOwnerDelivered(eventID, webhookID string) error {
return s.RecordDelivery(eventID, webhookID, DeliveryDelivered, "", 1)
}
// recordDeliveryHook runs inside RecordDelivery after its first statement,
// and dropDecisionHook just before a drop decision writes its row: tests
// interleave the two there. Both nil in production.
var (
recordDeliveryHook func()
dropDecisionHook func()
)
func nullIfEmpty(v string) any {
if v == "" {
return nil
}
return v
}
+42 -15
View File
@@ -64,6 +64,10 @@ type AppDelivery struct {
// event older than the hook's deliver_from.
OccurredAt string
Body []byte
// RateGate, when set, is asked before EVERY attempt, retries included,
// so a rate cap bounds requests rather than deliveries (codex r1 on
// U10c). false stops the delivery with AppRateLimited, nothing sent.
RateGate func() bool
}
// AppResult is what happened to one app delivery.
@@ -83,6 +87,9 @@ const (
// AppDeferred: admission could not be decided (a store error). Nothing
// was sent; the event is still owed.
AppDeferred
// AppRateLimited: the install's rate cap refused the attempt. Nothing
// was sent; the event is still owed and no attempt is charged.
AppRateLimited
)
// AppSignatureHeader carries `t=<unix seconds>,v1=<hex HMAC-SHA256 of
@@ -119,26 +126,34 @@ var newDeliveryID = func() string {
// with the owner path's retry schedule (maxDeliveryAttempts, linear backoff
// on context-aware timers). Each attempt is admitted separately: an install
// disabled between attempts gets no further attempt.
func (d *Dispatcher) DeliverAppEvent(adm AppAdmitter, poster AppPoster, dv AppDelivery) AppResult {
//
// It returns the final result and how many requests were actually sent, so
// the delivery ledger records real attempts (codex r2 on U10c).
func (d *Dispatcher) DeliverAppEvent(adm AppAdmitter, poster AppPoster, dv AppDelivery) (AppResult, int) {
parent := d.context()
result := AppPermanent
result, sent := AppPermanent, 0
for attempt := 1; attempt <= maxDeliveryAttempts; attempt++ {
result = d.attemptApp(parent, adm, poster, dv)
var s bool
result, s = d.attemptApp(parent, adm, poster, dv)
if s {
sent++
}
if result != AppTransient {
return result
return result, sent
}
if attempt < maxDeliveryAttempts {
if backoff := d.retryBackoff * time.Duration(attempt); backoff > 0 && !d.wait(parent, backoff) {
return AppTransient
return AppTransient, sent
}
}
}
return result
return result, sent
}
func (d *Dispatcher) attemptApp(parent context.Context, adm AppAdmitter, poster AppPoster, dv AppDelivery) AppResult {
// attemptApp makes one attempt and reports whether a request was sent.
func (d *Dispatcher) attemptApp(parent context.Context, adm AppAdmitter, poster AppPoster, dv AppDelivery) (AppResult, bool) {
if parent.Err() != nil {
return AppTransient
return AppTransient, false
}
// The deadline is RELATIVE and starts BEFORE admission (DOC-3371 §5):
// any delay before the send only shortens the attempt, while the
@@ -151,10 +166,10 @@ func (d *Dispatcher) attemptApp(parent context.Context, adm AppAdmitter, poster
a, err := adm.AdmitAppDelivery(dv.WebhookID, dv.Event, dv.CollectionID, dv.OccurredAt, deliveryID)
if err != nil {
if errors.Is(err, ErrAppDeliveryRefused) {
return AppRefused
return AppRefused, false
}
slog.Error("app webhook admission failed", "webhook_id", dv.WebhookID, "error", err)
return AppDeferred
return AppDeferred, false
}
defer func() {
if err := adm.EndAppDelivery(deliveryID); err != nil {
@@ -163,6 +178,13 @@ func (d *Dispatcher) attemptApp(parent context.Context, adm AppAdmitter, poster
}
}()
// The rate cap is asked AFTER admission: an attempt admission refuses
// sends nothing and must not spend the install's budget (codex r2 on
// U10c). The deferred EndAppDelivery releases the record.
if dv.RateGate != nil && !dv.RateGate() {
return AppRateLimited, false
}
h := http.Header{}
h.Set("Content-Type", "application/json")
h.Set("User-Agent", "Pad-Webhook/2.0")
@@ -171,22 +193,25 @@ func (d *Dispatcher) attemptApp(parent context.Context, adm AppAdmitter, poster
status, err := poster.Post(ctx, a.URL, dv.Body, h)
if err != nil {
// A request counts as sent only if the failure came after its bytes
// started going out (codex r3 on U10c).
sent := !errors.Is(err, appfetch.ErrRefused) && !errors.Is(err, appfetch.ErrNotSent)
if errors.Is(err, appfetch.ErrRefused) {
slog.Warn("app webhook refused by policy", "webhook_id", dv.WebhookID, "error", err)
return AppPermanent
return AppPermanent, sent
}
slog.Warn("app webhook delivery failed", "webhook_id", dv.WebhookID, "error", err)
return AppTransient
return AppTransient, sent
}
switch {
case status >= 200 && status < 300:
return AppDelivered
return AppDelivered, true
case status >= 500 && status < 600:
return AppTransient
return AppTransient, true
default:
// 3xx (never followed), 4xx, anything else.
slog.Warn("app webhook non-2xx", "webhook_id", dv.WebhookID, "status", status)
return AppPermanent
return AppPermanent, true
}
}
@@ -203,6 +228,8 @@ func (r AppResult) String() string {
return "refused"
case AppDeferred:
return "deferred"
case AppRateLimited:
return "rate_limited"
}
return fmt.Sprintf("AppResult(%d)", int(r))
}
+29 -6
View File
@@ -101,7 +101,7 @@ func TestAppDelivery_SignedAndAdmittedPerAttempt(t *testing.T) {
d, waits := newAppTestDispatcher()
adm := &fakeAdmitter{}
p := &fakePoster{statuses: []int{503, 502, 204}}
if got := d.DeliverAppEvent(adm, p, appDV); got != AppDelivered {
if got, _ := d.DeliverAppEvent(adm, p, appDV); got != AppDelivered {
t.Fatalf("result %s, want delivered", got)
}
if len(adm.admitted) != 3 || len(adm.ended) != 3 {
@@ -150,7 +150,7 @@ func TestAppDelivery_Outcomes(t *testing.T) {
} {
t.Run(tc.name, func(t *testing.T) {
d, _ := newAppTestDispatcher()
if got := d.DeliverAppEvent(tc.adm, tc.poster, appDV); got != tc.want {
if got, _ := d.DeliverAppEvent(tc.adm, tc.poster, appDV); got != tc.want {
t.Fatalf("result %s, want %s", got, tc.want)
}
if len(tc.poster.calls) != tc.attempts {
@@ -169,7 +169,7 @@ func TestAppDelivery_DisableBetweenAttemptsStopsTheRetry(t *testing.T) {
adm := &fakeAdmitter{}
adm.onAdmit = func(n int) { adm.refuse = n >= 1 }
p := &fakePoster{statuses: []int{503}}
if got := d.DeliverAppEvent(adm, p, appDV); got != AppRefused {
if got, _ := d.DeliverAppEvent(adm, p, appDV); got != AppRefused {
t.Fatalf("result %s, want refused", got)
}
if len(p.calls) != 1 {
@@ -189,7 +189,7 @@ func TestBackoff_StopEndsTheWait(t *testing.T) {
d.wait = timerWait
d.retryBackoff = time.Hour
start := time.Now()
if got := d.DeliverAppEvent(adm, p, appDV); got != AppTransient {
if got, _ := d.DeliverAppEvent(adm, p, appDV); got != AppTransient {
t.Fatalf("result %s, want transient (still owed)", got)
}
if len(p.calls) > 1 || time.Since(start) > 5*time.Second {
@@ -255,7 +255,7 @@ func TestAppDelivery_DeadlineStartsBeforeAdmission(t *testing.T) {
d, _ := newAppTestDispatcher()
adm := &fakeAdmitter{onAdmit: func(int) { time.Sleep(300 * time.Millisecond) }}
p := &fakePoster{}
if got := d.DeliverAppEvent(adm, p, appDV); got != AppDelivered {
if got, _ := d.DeliverAppEvent(adm, p, appDV); got != AppDelivered {
t.Fatalf("result %s", got)
}
if len(p.deadlines) != 1 || p.deadlines[0] > deliveryTimeout-250*time.Millisecond {
@@ -271,10 +271,33 @@ func TestAppDelivery_StoppedDispatcherSendsNothing(t *testing.T) {
d.SetContext(ctx)
adm := &fakeAdmitter{}
p := &fakePoster{}
if got := d.DeliverAppEvent(adm, p, appDV); got != AppTransient {
if got, _ := d.DeliverAppEvent(adm, p, appDV); got != AppTransient {
t.Fatalf("result %s, want transient (still owed)", got)
}
if len(adm.admitted) != 0 || len(p.calls) != 0 {
t.Fatalf("a stopped dispatcher admitted %d and sent %d", len(adm.admitted), len(p.calls))
}
}
// codex r3 on U10c: a request that failed before any byte went out (dial,
// handshake, policy) is not counted as sent.
func TestAppDelivery_UnsentFailuresAreNotCounted(t *testing.T) {
for _, tc := range []struct {
name string
err error
want AppResult
sent int
}{
{"dial failure", fmt.Errorf("%w: connection refused", appfetch.ErrNotSent), AppTransient, 0},
{"policy refusal", fmt.Errorf("%w: not https", appfetch.ErrRefused), AppPermanent, 0},
{"failed mid-request", errors.New("connection reset by peer"), AppTransient, 3},
} {
t.Run(tc.name, func(t *testing.T) {
d, _ := newAppTestDispatcher()
got, sent := d.DeliverAppEvent(&fakeAdmitter{}, &fakePoster{err: tc.err}, appDV)
if got != tc.want || sent != tc.sent {
t.Fatalf("got %s, %d sent; want %s, %d", got, sent, tc.want, tc.sent)
}
})
}
}
+31 -1
View File
@@ -79,6 +79,17 @@ type WebhookStore interface {
WorkspaceLive(workspaceID string) (bool, error)
}
// OwnerDeliveryLedger is optionally implemented by the store (TASK-3408
// U10c). When it is, DeliverEvent skips an owner hook that already received
// the event and records each owner success. It exists so that an event kept
// owed by someone ELSE (an app hook's rate limit or outage) is not delivered
// again to owner endpoints on every retry. Owner failure, retry and ack
// semantics are unchanged; full owner per-endpoint state is TASK-3409.
type OwnerDeliveryLedger interface {
OwnerDelivered(eventID, webhookID string) (bool, error)
RecordOwnerDelivered(eventID, webhookID string) error
}
// errWorkspaceGone stops a request whose webhook's workspace was deleted while
// it was in flight; errWorkspaceUnknown stops one whose workspace could not be
// checked (BUG-3340). The first is suppressed, the second deferred.
@@ -410,7 +421,26 @@ func (d *Dispatcher) DeliverEvent(dv Delivery) (DeliveryOutcome, error) {
continue
}
out.Matched++
switch d.deliver(hook, body) {
ledger, _ := d.store.(OwnerDeliveryLedger)
if ledger != nil && dv.EventID != "" {
done, err := ledger.OwnerDelivered(dv.EventID, hook.ID)
if err != nil {
return out, fmt.Errorf("owner delivery ledger: %w", err)
}
if done {
out.Succeeded++ // received on an earlier pass
continue
}
}
res := d.deliver(hook, body)
if res == deliverySuccess && ledger != nil && dv.EventID != "" {
if err := ledger.RecordOwnerDelivered(dv.EventID, hook.ID); err != nil {
// Not fatal: the cost is one possible re-send, the
// behaviour before the ledger existed.
slog.Error("owner delivery ledger: record failed", "webhook_id", hook.ID, "error", err)
}
}
switch res {
case deliverySuppressed:
// The workspace was deleted mid-delivery: this hook no longer
// selects the event, and nothing is owed to it.