diff --git a/internal/appfetch/appfetch.go b/internal/appfetch/appfetch.go index 1103a25f5..fed67c06b 100644 --- a/internal/appfetch/appfetch.go +++ b/internal/appfetch/appfetch.go @@ -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 +} diff --git a/internal/appfetch/poster_test.go b/internal/appfetch/poster_test.go index 285c7f933..c65beadab 100644 --- a/internal/appfetch/poster_test.go +++ b/internal/appfetch/poster_test.go @@ -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") + } +} diff --git a/internal/server/app_webhook_delivery.go b/internal/server/app_webhook_delivery.go index e46f4fbea..f18991c09 100644 --- a/internal/server/app_webhook_delivery.go +++ b/internal/server/app_webhook_delivery.go @@ -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) + } + } +} diff --git a/internal/server/app_webhooks.go b/internal/server/app_webhooks.go index 279094347..83370f758 100644 --- a/internal/server/app_webhooks.go +++ b/internal/server/app_webhooks.go @@ -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 } diff --git a/internal/server/handlers_apps_u10c_test.go b/internal/server/handlers_apps_u10c_test.go new file mode 100644 index 000000000..aa9f40ae3 --- /dev/null +++ b/internal/server/handlers_apps_u10c_test.go @@ -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) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index eb6a2b659..4233f19f5 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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 diff --git a/internal/store/app_delivery.go b/internal/store/app_delivery.go index 42fa9e99e..4b79261ad 100644 --- a/internal/store/app_delivery.go +++ b/internal/store/app_delivery.go @@ -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:" 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. diff --git a/internal/store/app_webhooks.go b/internal/store/app_webhooks.go index 9c437374c..eccb16db7 100644 --- a/internal/store/app_webhooks.go +++ b/internal/store/app_webhooks.go @@ -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 } diff --git a/internal/store/migrations/124_webhook_deliveries.sql b/internal/store/migrations/124_webhook_deliveries.sql new file mode 100644 index 000000000..90b3b4031 --- /dev/null +++ b/internal/store/migrations/124_webhook_deliveries.sql @@ -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; diff --git a/internal/store/nul_unprotected_baseline.txt b/internal/store/nul_unprotected_baseline.txt index 0a64b760a..90d9017a5 100644 --- a/internal/store/nul_unprotected_baseline.txt +++ b/internal/store/nul_unprotected_baseline.txt @@ -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 diff --git a/internal/store/pgmigrations/098_webhook_deliveries.sql b/internal/store/pgmigrations/098_webhook_deliveries.sql new file mode 100644 index 000000000..27d16c8f8 --- /dev/null +++ b/internal/store/pgmigrations/098_webhook_deliveries.sql @@ -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; diff --git a/internal/store/task3408_ledger_test.go b/internal/store/task3408_ledger_test.go new file mode 100644 index 000000000..13260f33e --- /dev/null +++ b/internal/store/task3408_ledger_test.go @@ -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) + } +} diff --git a/internal/store/webhook_deliveries.go b/internal/store/webhook_deliveries.go new file mode 100644 index 000000000..12007dd8e --- /dev/null +++ b/internal/store/webhook_deliveries.go @@ -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 +} diff --git a/internal/webhooks/app_delivery.go b/internal/webhooks/app_delivery.go index 06c2487d2..1b5873eb3 100644 --- a/internal/webhooks/app_delivery.go +++ b/internal/webhooks/app_delivery.go @@ -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=,v1= 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)) } diff --git a/internal/webhooks/app_delivery_test.go b/internal/webhooks/app_delivery_test.go index 06b8daa80..5bca1f927 100644 --- a/internal/webhooks/app_delivery_test.go +++ b/internal/webhooks/app_delivery_test.go @@ -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) + } + }) + } +} diff --git a/internal/webhooks/dispatcher.go b/internal/webhooks/dispatcher.go index b8bd8337a..3b95383ad 100644 --- a/internal/webhooks/dispatcher.go +++ b/internal/webhooks/dispatcher.go @@ -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.