diff --git a/cmd/pad/cmd_push.go b/cmd/pad/cmd_push.go index 23d3b3456..d158d9685 100644 --- a/cmd/pad/cmd_push.go +++ b/cmd/pad/cmd_push.go @@ -3,6 +3,7 @@ package main import ( "errors" "fmt" + "github.com/google/uuid" "os" "strings" @@ -30,7 +31,7 @@ const maxPushMessageLenForHelp = 4096 // notification-etiquette section for the receiving-agent behavior // contract. func pushCmd() *cobra.Command { - var message string + var message, pushOpKey string cmd := &cobra.Command{ Use: "push ", @@ -75,7 +76,13 @@ func pushCmd() *cobra.Command { client, _ := getClient() ws := getWorkspace() - result, err := client.PushItem(ws, args[0], trimmed) + // TASK-3585: one idempotency key per invocation, so the + // client's single retry, or a re-run with --op-key, does not + // deliver the instruction twice. + if pushOpKey == "" { + pushOpKey = uuid.NewString() + } + result, err := client.PushItemKeyed(ws, args[0], trimmed, pushOpKey) if err != nil { // AMBIGUOUS FAILURES GET SAID OUT LOUD (codex round 9 on // BUG-2699). The handler publishes BEFORE it writes its @@ -83,9 +90,10 @@ func pushCmd() *cobra.Command { // nowhere: a response lost in transit, a truncated body, a // gateway envelope, or the server's own push_unconfirmed all // reach here identically to a clean refusal. A user who reads - // "error" and re-runs the command delivers the push twice — - // there is no idempotency key, and the receiving agent acts - // on each copy. + // "error" and re-runs the command delivers the push twice: + // the re-run gets a NEW key (TASK-3585 keys one invocation, + // and the client already made its one keyed retry), and the + // receiving agent acts on each copy. // // The web dialog has drawn this distinction since TASK-2588; // the CLI had not, so the same outcome produced a safe @@ -106,6 +114,7 @@ func pushCmd() *cobra.Command { }, } + cmd.Flags().StringVar(&pushOpKey, "op-key", "", "idempotency key: re-running the SAME push with the same key answers the first push instead of delivering it again (default: a new key per invocation)") cmd.Flags().StringVarP(&message, "message", "m", "", fmt.Sprintf("instruction text to push (required, max %d characters after whitespace collapse)", maxPushMessageLenForHelp)) return cmd diff --git a/cmd/pad/cmd_watch.go b/cmd/pad/cmd_watch.go index e268dc873..322cf2289 100644 --- a/cmd/pad/cmd_watch.go +++ b/cmd/pad/cmd_watch.go @@ -11,6 +11,7 @@ import ( "os/signal" "path/filepath" "strings" + "sync" "syscall" "text/tabwriter" "time" @@ -67,8 +68,56 @@ type watchStreamPayload struct { Kind string `json:"kind"` Actor string `json:"actor"` Summary string `json:"summary"` + // PushID is set on a push (TASK-3585). A re-publish of the same push + // (a retry after a server crash) carries the same id; seenPushes drops it. + PushID string `json:"push_id,omitempty"` } +// recentSet remembers ids for a bounded time and count, so the monitor can +// drop a push it has already printed (TASK-3585). A monitor that predates +// push_id prints every publish: at-least-once, as before. +type recentSet struct { + mu sync.Mutex + max int + ttl time.Duration + now func() time.Time + order []string + at map[string]time.Time +} + +func newRecentSet(max int, ttl time.Duration) *recentSet { + return &recentSet{max: max, ttl: ttl, now: time.Now, at: map[string]time.Time{}} +} + +// seen reports whether id was recorded within the window, and records it +// when it was not. +func (s *recentSet) seen(id string) bool { + s.mu.Lock() + defer s.mu.Unlock() + now := s.now() + if t, ok := s.at[id]; ok && now.Sub(t) <= s.ttl { + return true + } + // Evict from the oldest end: anything past the ttl, and enough to make + // room for this one under max. An expired copy of id goes here too. + for len(s.order) > 0 { + oldest := s.order[0] + if len(s.order) < s.max && now.Sub(s.at[oldest]) <= s.ttl && oldest != id { + break + } + delete(s.at, oldest) + s.order = s.order[1:] + } + s.at[id] = now + s.order = append(s.order, id) + return false +} + +// seenPushes is the monitor's memory of printed pushes: about the last 256, +// for ten minutes. One monitor runs per process, so one set serves every +// reconnect. +var seenPushes = newRecentSet(256, 10*time.Minute) + // formatMonitorLine renders one notification as the exact stdout-line // contract `pad watch --stream --for-session` promises (DOC-2479): // "PAD demo/TASK-214 → done (Dave): fix verified" — one line, facts @@ -622,7 +671,8 @@ func streamWatchEventsTo(w io.Writer, resp *http.Response, lastEventID string) s } if data != "" { var payload watchStreamPayload - if err := json.Unmarshal([]byte(data), &payload); err == nil { + if err := json.Unmarshal([]byte(data), &payload); err == nil && + (payload.PushID == "" || !seenPushes.seen(payload.PushID)) { fmt.Fprintln(w, formatMonitorLine(payload)) resyncAnnounced = false } diff --git a/cmd/pad/task3585_monitor_dedupe_test.go b/cmd/pad/task3585_monitor_dedupe_test.go new file mode 100644 index 000000000..5a458f116 --- /dev/null +++ b/cmd/pad/task3585_monitor_dedupe_test.go @@ -0,0 +1,50 @@ +package main + +import ( + "bytes" + "strings" + "testing" + "time" +) + +// TASK-3585: the monitor prints a push once, however many times it is +// published under the same push_id (a re-publish after a claim takeover). +func TestTASK3585_MonitorDropsARepublishedPush(t *testing.T) { + prev := seenPushes + seenPushes = newRecentSet(256, 10*time.Minute) + t.Cleanup(func() { seenPushes = prev }) + armSessionForTest(t) + + ev := func(id, pushID string) string { + return "id: " + id + "\nevent: notification\ndata: {\"id\":" + id + + ",\"workspace\":\"w\",\"item_ref\":\"TASK-1\",\"kind\":\"push\",\"summary\":\"triage this\",\"push_id\":\"" + pushID + "\"}\n\n" + } + var buf bytes.Buffer + streamWatchEventsTo(&buf, fakeSSEResponse(ev("1", "push_a")+ev("2", "push_a")+ev("3", "push_b")), "") + if n := strings.Count(buf.String(), "triage this"); n != 2 { + t.Fatalf("want push_a once and push_b once (2 lines), got %d:\n%s", n, buf.String()) + } + // The set outlives one connection: a reconnect replaying push_a is dropped too. + buf.Reset() + streamWatchEventsTo(&buf, fakeSSEResponse(ev("4", "push_a")), "3") + if buf.Len() != 0 { + t.Fatalf("a re-publish on a new connection was printed: %q", buf.String()) + } +} + +func TestTASK3585_RecentSetBounds(t *testing.T) { + now := time.Unix(0, 0) + s := newRecentSet(2, time.Minute) + s.now = func() time.Time { return now } + if s.seen("a") || s.seen("b") || !s.seen("a") { + t.Fatal("a and b are new; a again is a repeat") + } + s.seen("c") // over max: evicts the oldest (a) + if s.seen("a") { + t.Fatal("a should have been evicted by count") + } + now = now.Add(2 * time.Minute) + if s.seen("c") { + t.Fatal("c should have expired") + } +} diff --git a/docs/api.md b/docs/api.md index fe2295d4a..f2ec3734f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -72,7 +72,8 @@ REST API at `/api/v1/`. Key endpoints: - `GET/POST /workspaces/{ws}/items/{slug}/links` — item relationships (blocks/blocked-by, parent/child) - `GET /search?q=query&workspace=slug` — full-text search. A `collection` filter resolves like a collection in the item routes (BUG-2659): exact slug first, then the singular/alias fallback, and an archived collection still claims its name. It resolves once per workspace in scope, so `collection=task` means `task` where one exists and `tasks` where it does not. `GET /server/capabilities` advertises it as `search_collection_resolution`; the CLI sends the name as typed only to a server that does, and expands the shorthand itself otherwise. `include_archived=true` (TASK-2864) adds soft-deleted items on every path (full text, ref and number lookups, total, facets), each carrying `deleted_at`. The permission filter is unchanged, and an individual item grant names live items only, so a guest sees an archived item only through a collection grant. The web search box's `is:archived` token sends it; not on the CLI or MCP - `GET /api/v1/events?workspace=slug` — SSE real-time events (workspace-scoped) -- `GET /api/v1/events/stream` — SSE watch/push notifications (USER-scoped, spans every workspace the caller belongs to; backs `pad watch --stream`). With `?access=true` (TASK-3272) the stream also carries `workspace_access_changed` addressed to the caller — `{workspace_id, change}` with change `gained` / `lost` / `deleted` / `restored` / `purged` — published after every door that changes whether an EXISTING account reaches a workspace (signup-time doors publish nothing: the new account holds no stream yet). Opt-in because an installed `pad watch` prints any kind it does not know as an item line into an agent session. It is a refetch hint: correctness lives on the read-filtered `GET /me/workspace-tabs`, and a duplicate or missed event under concurrent grant writes is benign. Such a stream counts against `PAD_SSE_MAX_PER_USER` like any other +- `GET /api/v1/events/stream` — SSE watch/push notifications (USER-scoped, spans every workspace the caller belongs to; backs `pad watch --stream`). With `?access=true` (TASK-3272) the stream also carries `workspace_access_changed` addressed to the caller — `{workspace_id, change}` with change `gained` / `lost` / `deleted` / `restored` / `purged` — published after every door that changes whether an EXISTING account reaches a workspace (signup-time doors publish nothing: the new account holds no stream yet). Opt-in because an installed `pad watch` prints any kind it does not know as an item line into an agent session. It is a refetch hint: correctness lives on the read-filtered `GET /me/workspace-tabs`, and a duplicate or missed event under concurrent grant writes is benign. Such a stream counts against `PAD_SSE_MAX_PER_USER` like any other. A push event carries `push_id` (TASK-3585, omitted on every other kind): the same id on every publish of one keyed push, so a consumer drops one it has seen (`pad watch --stream --for-session` keeps ~256 ids for 10 minutes); a monitor that predates it is at-least-once +- `POST /workspaces/{ws}/items/{ref}/push` — idempotency key (TASK-3585): `op_key` in the body or `Idempotency-Key`, claimed after every refusal. Push writes no row, so the key is completed AFTER the publish, outside any transaction: a clean publish stores the response and a same-key retry replays it without publishing; an unconfirmed publish (`push_unconfirmed`, 502) stores 409 `push_outcome_unknown` for a retry, which publishes nothing; a closed bus (503) releases the key. A crash between publish and completion lets the 60s lease expire and a retry publish again, under the same `push_id`, which is what a monitor dedupes on. The CLI keys every push and retries once after a 5xx, only against a server advertising `idempotency_keys` **Every long-lived connection ends when the CREDENTIAL that opened it stops being valid** — both SSE endpoints and the `/api/v1/collab/{itemID}` WebSocket, checked on each connection's existing revalidation tick (BUG-3007). The invariant is deliberately wider than "the session was destroyed": a revoked PAT ends a stream too, because a revoked PAT still streaming is the same defect for a CLI or MCP caller that a destroyed session is for a browser. Before this, all three kept running indefinitely — measured at 150-180s past a logout, and 64-120s past a revocation, in both cases with `/auth/me` on that credential answering `401`. The pre-existing per-tick checks are about the PRINCIPAL (is this user still a member, is the item still visible) and none of them changes when a credential dies, which is why the two are separate questions rather than one. A request carrying NO credential AND no resolved principal — the fresh-install window, the legacy no-auth path — is unaffected: there is nothing to invalidate. A request that carries a resolved USER but no record of how the credential was established is CLOSED, not exempted: the middleware stamps that on every accept point, so its absence means an accept point recorded who without recording what, and a connection nobody can re-check is the same defect under a new name. (Both MCP accept points are in that state by design today — the PAT one records `api_token` and is re-checked; the OAuth one has no liveness door for an opaque token yet, so it fails closed. Neither is reachable: no MCP route is long-lived.) A store ERROR is not a revocation and keeps the connection until the next tick, so a database blip does not disconnect the fleet. diff --git a/internal/cli/client.go b/internal/cli/client.go index 4021eb349..4ac788e60 100644 --- a/internal/cli/client.go +++ b/internal/cli/client.go @@ -641,9 +641,27 @@ type PushResult struct { // --stream --for-session` consumes. Transient, fire-and-forget — see // server.handlePushToItem's doc comment for the no-durability rationale. func (c *Client) PushItem(wsSlug, itemSlug, message string) (*PushResult, error) { + return c.PushItemKeyed(wsSlug, itemSlug, message, "") +} + +// PushItemKeyed is PushItem with an idempotency key (TASK-3585). With a key +// it retries once with that key after a transport error or a 5xx, and only +// against a server that advertises idempotency_keys: there the retry +// answers the first push (or 409 push_outcome_unknown when that push could +// not be confirmed) instead of delivering it again. +func (c *Client) PushItemKeyed(wsSlug, itemSlug, message, opKey string) (*PushResult, error) { body := map[string]string{"message": message} + if opKey != "" { + body["op_key"] = opKey + } + path := "/workspaces/" + wsSlug + "/items/" + itemSlug + "/push" var result PushResult - return &result, c.post("/workspaces/"+wsSlug+"/items/"+itemSlug+"/push", body, &result) + err := c.post(path, body, &result) + if err != nil && opKey != "" && retryableWithKey(err) && c.serverHonorsKeys() { + result = PushResult{} + err = c.post(path, body, &result) + } + return &result, err } // ListWatches returns every watch the current user holds, across all diff --git a/internal/cli/task3585_push_retry_test.go b/internal/cli/task3585_push_retry_test.go new file mode 100644 index 000000000..cec92c414 --- /dev/null +++ b/internal/cli/task3585_push_retry_test.go @@ -0,0 +1,69 @@ +package cli + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +// TASK-3585: a keyed push retries once with the same key after a 5xx, and +// only against a server that advertises idempotency_keys. +func pushKeyServer(t *testing.T, honorsKeys bool, statuses ...int) (*Client, func() []string) { + t.Helper() + var mu sync.Mutex + var keys []string + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/capabilities") { + _ = json.NewEncoder(w).Encode(map[string]any{"idempotency_keys": honorsKeys}) + return + } + raw, _ := io.ReadAll(r.Body) + var b map[string]any + _ = json.Unmarshal(raw, &b) + k, _ := b["op_key"].(string) + mu.Lock() + n := len(keys) + keys = append(keys, k) + mu.Unlock() + status := http.StatusOK + if n < len(statuses) { + status = statuses[n] + } + w.WriteHeader(status) + if status >= 400 { + _, _ = w.Write([]byte(`{"error":{"code":"push_unconfirmed","message":"m"}}`)) + return + } + _, _ = w.Write([]byte(`{"ref":"TASK-1","workspace":"w","pushed":true,"message":"m"}`)) + })) + t.Cleanup(ts.Close) + t.Setenv("HOME", t.TempDir()) + return NewClientFromURL(ts.URL), func() []string { mu.Lock(); defer mu.Unlock(); return append([]string(nil), keys...) } +} + +func TestTASK3585_PushRetriesOnceWithTheSameKey(t *testing.T) { + c, keys := pushKeyServer(t, true, http.StatusBadGateway) + if _, err := c.PushItemKeyed("w", "TASK-1", "m", "k-1"); err != nil { + t.Fatalf("a 502 then a 200 must succeed: %v", err) + } + if got := keys(); len(got) != 2 || got[0] != "k-1" || got[1] != "k-1" { + t.Fatalf("want two attempts with one key; got %v", got) + } +} + +func TestTASK3585_PushNoRetryWithoutTheCapabilityOrAKey(t *testing.T) { + c, keys := pushKeyServer(t, false, http.StatusBadGateway) + _, _ = c.PushItemKeyed("w", "TASK-1", "m", "k-2") + if got := keys(); len(got) != 1 { + t.Fatalf("retried a push against a server that ignores keys: %d attempts", len(got)) + } + c2, keys2 := pushKeyServer(t, true, http.StatusBadGateway) + _, _ = c2.PushItem("w", "TASK-1", "m") + if got := keys2(); len(got) != 1 { + t.Fatalf("retried a keyless push: %d attempts", len(got)) + } +} diff --git a/internal/server/handlers_push.go b/internal/server/handlers_push.go index 216d67016..b6c2f9124 100644 --- a/internal/server/handlers_push.go +++ b/internal/server/handlers_push.go @@ -30,6 +30,9 @@ type pushRequest struct { // silently dropping a field this response documents at length. It // still never SENDS target_session_id. TargetSessionID string `json:"target_session_id,omitempty"` + // OpKey is the push's idempotency key in the body (TASK-3585). Not part + // of the request hash. + OpKey string `json:"op_key,omitempty"` } // pushResponse is the body of a successful push (dispatcher review round @@ -333,6 +336,31 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) { "item_ref", item.Ref, "error", presenceErr) deliveredSessionsUnknown = true } + + // TASK-3585: an idempotency key, claimed after every refusal above and + // before the publish. Push writes no row, so its answer cannot commit + // with it: the key is completed AFTER the publish, outside any + // transaction. A crash in between lets the lease expire and a retry + // publish again, which is why every publish carries a push_id derived + // from the key: a monitor that has printed it drops the repeat. + key, kerr := idempotencyKey(r, input.OpKey) + if kerr != nil { + writeError(w, http.StatusBadRequest, "bad_request", kerr.Error()) + return + } + hold, ok := s.claimKey(w, r, workspaceID, idempotencyRoutePush, item.ID, + pushRequest{Message: message, TargetSessionID: targetSessionID}, key, + func(status int, stored string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(stored)) + }) + if !ok { + return + } + defer s.release(hold) + pushID := pushIDFor(workspaceID, userID, key) + if targetSessionID == "" || deliveredSessions > 0 { // The ONE direct Bus.Publish call in this package, and the only // one that acts on the result (BUG-2699 — see @@ -346,6 +374,7 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) { // instruction is gone, and the caller is the only one who can do // anything about that. if err := s.watchEvents.Publish(watchevents.Notification{ + PushID: pushID, WorkspaceID: workspaceID, ItemID: item.ID, CollectionID: item.CollectionID, @@ -357,6 +386,13 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) { TargetUserID: userID, TargetSessionID: targetSessionID, }); err != nil { + // An unconfirmed publish may have delivered: the key records + // that, so a same-key retry answers 409 push_outcome_unknown + // instead of delivering a second time. A closed bus published + // nothing, and the deferred release frees the key. + if hold != nil && !errors.Is(err, watchevents.ErrBusClosed) { + s.completeKey(hold, http.StatusConflict, pushOutcomeUnknownEnvelope(item.Ref)) + } writePushPublishError(w, err, item.Ref) return } @@ -371,6 +407,9 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) { if !deliveredSessionsUnknown { resp.DeliveredSessions = &deliveredSessions } + if hold != nil { + s.completeKey(hold, http.StatusOK, resp) + } writeJSON(w, http.StatusOK, resp) } diff --git a/internal/server/handlers_watch_events.go b/internal/server/handlers_watch_events.go index 8c5a9b84c..730c69b6a 100644 --- a/internal/server/handlers_watch_events.go +++ b/internal/server/handlers_watch_events.go @@ -56,6 +56,9 @@ type watchEventPayload struct { // may no longer resolve (purged), where Workspace falls back to the ID. WorkspaceID string `json:"workspace_id,omitempty"` Change string `json:"change,omitempty"` + // PushID is set only on a push (TASK-3585) and omitted otherwise: a + // monitor that has already printed this push_id drops the event. + PushID string `json:"push_id,omitempty"` } // handleWatchEventsStream streams the caller's watch/nudge notifications. @@ -862,6 +865,7 @@ func watchEventPayloadFor(s *Server, n watchevents.Notification) watchEventPaylo Kind: n.Kind, Actor: actor, Summary: n.Summary, + PushID: n.PushID, } if n.Kind == watchevents.KindWorkspaceAccessChanged { p.WorkspaceID = n.WorkspaceID diff --git a/internal/server/idempotency_keys.go b/internal/server/idempotency_keys.go index 9a4d647f5..e06594c85 100644 --- a/internal/server/idempotency_keys.go +++ b/internal/server/idempotency_keys.go @@ -6,8 +6,11 @@ import ( "encoding/hex" "encoding/json" "fmt" + "log/slog" "net/http" + "github.com/google/uuid" + "github.com/PerpetualSoftware/pad/internal/models" "github.com/PerpetualSoftware/pad/internal/store" ) @@ -51,10 +54,8 @@ func (s *Server) claimKey(w http.ResponseWriter, r *http.Request, workspaceID, r if key == "" { return nil, true } - body, _ := json.Marshal(canon) - sum := sha256.Sum256([]byte(route + "\x00" + scope + "\x00" + string(body))) userID := currentUserID(r) - c, err := s.store.ClaimIdempotencyKey(workspaceID, userID, key, route, hex.EncodeToString(sum[:]), idempotencyLease) + c, err := s.store.ClaimIdempotencyKey(workspaceID, userID, key, route, idemRequestHash(route, scope, canon), idempotencyLease) if err != nil { writeInternalError(w, err) return nil, false @@ -149,3 +150,54 @@ func (s *Server) replayAppend(w http.ResponseWriter, r *http.Request, visIDs []s // request hash. Its answer is the whole first ItemCopyResult: the warnings // it carries (dropped fields, attachments cloned) cannot be recomputed later. const idempotencyRouteCopy = "item.copy" + +// idempotencyRoutePush names the push route in a key's request hash. +const idempotencyRoutePush = "item.push" + +// pushOutcomeUnknownCode answers a same-key retry of a push whose publish +// could not be confirmed (TASK-3585): it may have been delivered, so it is +// not published again. +const pushOutcomeUnknownCode = "push_outcome_unknown" + +// pushOutcomeUnknownEnvelope is the stored answer for that retry, in the +// API's error envelope. +func pushOutcomeUnknownEnvelope(ref string) any { + return map[string]any{"error": map[string]any{ + "code": pushOutcomeUnknownCode, + "message": "An earlier push with this key could not be confirmed; it may or may not have been delivered, so it is not sent again. " + + "Check your agent session; to send it anyway, use a NEW key.", + "details": map[string]any{"ref": ref}, + }} +} + +// completeKey records an answer for a write that has no transaction of its +// own (push). A failure is logged, not returned: the write already +// happened, and the push_id still lets a monitor drop a re-publish. +func (s *Server) completeKey(h *idemHold, status int, answer any) { + b, err := json.Marshal(answer) + if err == nil { + err = s.store.CompleteIdempotencyKey(h.workspaceID, h.userID, h.key, h.token, status, string(b)) + } + if err != nil { + slog.Warn("idempotency key not completed after a write with no transaction", "key", h.key, "error", err) + } +} + +// pushIDFor is the push_id a publish carries (TASK-3585). With a key it is +// derived from (workspace, user, key), so a re-publish after a claim +// takeover carries the same id; without one it is fresh. +func pushIDFor(workspaceID, userID, key string) string { + if key == "" { + return uuid.NewString() + } + sum := sha256.Sum256([]byte(workspaceID + "\x00" + userID + "\x00" + key)) + return "push_" + hex.EncodeToString(sum[:16]) +} + +// idemRequestHash is the request half of a key's identity: the route, its +// scope (the item) and the canonical request body. +func idemRequestHash(route, scope string, canon any) string { + body, _ := json.Marshal(canon) + sum := sha256.Sum256([]byte(route + "\x00" + scope + "\x00" + string(body))) + return hex.EncodeToString(sum[:]) +} diff --git a/internal/server/task3585_push_key_test.go b/internal/server/task3585_push_key_test.go new file mode 100644 index 000000000..356f36397 --- /dev/null +++ b/internal/server/task3585_push_key_test.go @@ -0,0 +1,141 @@ +package server + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "sync" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/watchevents" +) + +// TASK-3585: idempotency keys on push. Push writes no row, so the key is +// completed after the publish; every publish carries a push_id derived from +// the key so a monitor can drop a re-publish after a claim takeover. + +// capturingBus records what was published and fails with err when set. +type capturingBus struct { + mu sync.Mutex + err error + sent []watchevents.Notification +} + +func (b *capturingBus) Publish(n watchevents.Notification) error { + b.mu.Lock() + defer b.mu.Unlock() + b.sent = append(b.sent, n) + return b.err +} +func (b *capturingBus) published() []watchevents.Notification { + b.mu.Lock() + defer b.mu.Unlock() + return append([]watchevents.Notification(nil), b.sent...) +} +func (b *capturingBus) setErr(err error) { b.mu.Lock(); b.err = err; b.mu.Unlock() } +func (b *capturingBus) Subscribe() (chan watchevents.Notification, <-chan struct{}, error) { + return nil, nil, nil +} +func (b *capturingBus) SubscribeAndReplaySince(context.Context, int64) (chan watchevents.Notification, []watchevents.Notification, <-chan struct{}, error) { + return nil, nil, nil, nil +} +func (b *capturingBus) Unsubscribe(chan watchevents.Notification) {} +func (b *capturingBus) EventsSince(int64) []watchevents.Notification { + return nil +} +func (b *capturingBus) Close() {} + +func pushOnce(t *testing.T, srv *Server, slug, itemSlug, token, key string) *http.Response { + t.Helper() + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+itemSlug+"/push", token, + map[string]interface{}{"message": "triage this", "op_key": key}) + return rr.Result() +} + +func TestTASK3585_PushReplaysWithoutRedelivering(t *testing.T) { + srv := testServer(t) + bus := &capturingBus{} + srv.SetWatchEventsBus(bus) + slug, item, tok, _ := setupWatchTestUser(t, srv) + + first := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-1") + again := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-1") + if first.StatusCode != http.StatusOK || again.StatusCode != http.StatusOK || again.Header.Get("Idempotent-Replayed") != "true" { + t.Fatalf("replay: %d/%d", first.StatusCode, again.StatusCode) + } + if n := len(bus.published()); n != 1 { + t.Fatalf("a same-key retry published again: %d publishes", n) + } + if bus.published()[0].PushID == "" { + t.Fatal("a push must carry a push_id") + } +} + +func TestTASK3585_AmbiguousPushIsNotRedelivered(t *testing.T) { + srv := testServer(t) + bus := &capturingBus{err: errors.New("redis: i/o timeout")} + srv.SetWatchEventsBus(bus) + slug, item, tok, _ := setupWatchTestUser(t, srv) + + first := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-2") + if first.StatusCode != http.StatusBadGateway { + t.Fatalf("premise: an unconfirmed publish answers 502; got %d", first.StatusCode) + } + bus.setErr(nil) + again := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-2") + var env struct { + Error struct{ Code string } `json:"error"` + } + _ = json.NewDecoder(again.Body).Decode(&env) + if again.StatusCode != http.StatusConflict || env.Error.Code != pushOutcomeUnknownCode { + t.Fatalf("retry after an ambiguous publish: %d %q", again.StatusCode, env.Error.Code) + } + if n := len(bus.published()); n != 1 { + t.Fatalf("the ambiguous push was delivered again: %d publishes", n) + } +} + +func TestTASK3585_ClosedBusReleasesThePushKey(t *testing.T) { + srv := testServer(t) + bus := &capturingBus{err: watchevents.ErrBusClosed} + srv.SetWatchEventsBus(bus) + slug, item, tok, _ := setupWatchTestUser(t, srv) + + if r := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-3"); r.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("premise: a closed bus answers 503; got %d", r.StatusCode) + } + bus.setErr(nil) + again := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-3") + if again.StatusCode != http.StatusOK || again.Header.Get("Idempotent-Replayed") != "" { + t.Fatalf("a push that published nothing must be sendable again with its key: %d", again.StatusCode) + } +} + +// A server that dies between publishing and completing the key leaves a +// claim whose lease expires. The retry takes it over and publishes again, +// with the SAME push_id, which is what lets a monitor drop the repeat. +func TestTASK3585_PushTakeoverKeepsThePushID(t *testing.T) { + srv := testServer(t) + bus := &capturingBus{} + srv.SetWatchEventsBus(bus) + slug, item, tok, user := setupWatchTestUser(t, srv) + ws, err := srv.store.GetWorkspaceBySlug(slug) + if err != nil || ws == nil { + t.Fatalf("workspace: %v", err) + } + hash := idemRequestHash(idempotencyRoutePush, item.ID, pushRequest{Message: "triage this"}) + // The dead first attempt: claimed, never completed, lease already over. + if _, err := srv.store.ClaimIdempotencyKey(ws.ID, user.ID, "p-4", idempotencyRoutePush, hash, -time.Second); err != nil { + t.Fatal(err) + } + r := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-4") + if r.StatusCode != http.StatusOK { + t.Fatalf("takeover: %d", r.StatusCode) + } + sent := bus.published() + if len(sent) != 1 || sent[0].PushID != pushIDFor(ws.ID, user.ID, "p-4") { + t.Fatalf("the takeover must publish the key's stable push_id: %+v", sent) + } +} diff --git a/internal/store/idempotency.go b/internal/store/idempotency.go index fd0c46609..a2d405e81 100644 --- a/internal/store/idempotency.go +++ b/internal/store/idempotency.go @@ -119,6 +119,20 @@ func (s *Store) CompleteIdempotencyKeyTx(tx *sql.Tx, workspaceID, userID, key, t return nil } +// CompleteIdempotencyKey is CompleteIdempotencyKeyTx for a write that has +// no transaction of its own (push, TASK-3585), in a transaction of its own. +func (s *Store) CompleteIdempotencyKey(workspaceID, userID, key, token string, status int, response string) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if err := s.CompleteIdempotencyKeyTx(tx, workspaceID, userID, key, token, status, response); err != nil { + return err + } + return tx.Commit() +} + // ReleaseIdempotencyKey gives up a claim the write refused, so a corrected // retry with the same key can proceed (lead ruling: refusals are not // stored). A pending claim is deleted; an unknown one keeps its record and diff --git a/internal/watchevents/watchevents.go b/internal/watchevents/watchevents.go index 086ff58c6..db6685a99 100644 --- a/internal/watchevents/watchevents.go +++ b/internal/watchevents/watchevents.go @@ -145,7 +145,12 @@ const ( type Notification struct { // ID is a monotonic sequence number assigned by the Bus on Publish, // used for Last-Event-ID resume. Zero until Publish assigns it. - ID int64 + ID int64 + // PushID identifies one push across re-publishes (TASK-3585): for a + // keyed push it is derived from the key, so a re-publish after a claim + // takeover carries the same id and a monitor drops the repeat. Empty on + // every other kind. + PushID string WorkspaceID string ItemID string // CollectionID is the watched/assigned item's collection, needed so