diff --git a/API.md b/API.md index 5703264..dbc3c5d 100644 --- a/API.md +++ b/API.md @@ -646,15 +646,15 @@ connect (see "Reconnect catch-up"), which reaches the native service even when the JS layer is not running. The 409 is the secondary path for a client that is running and polling but has no socket — both settle the same rows. -- The rows are marked acknowledged by the act of returning this response, so the - same uninterrupted alarm cannot produce a second `409`. -- On the live path the WebSocket `ack` frame marks them too, so an online client - that actually received the frame is never told again. +- The rows are marked **delivered** (`delivered_at`) by the act of returning this + response, so the same uninterrupted alarm cannot produce a second `409`. That + is a receipt, not a dismissal: it does not end the alert. +- On the live path the WebSocket `ack` frame records the same receipt, so an + online client that actually received the frame is never told again. - Everything is idempotent: an ack for an event the device never owed, or a repeated ack, updates nothing and is not an error. - A trigger that was already disarmed before the device checked back in is not - news: the ledger entries are cleared and the normal body (`active: false`) is - returned instead of a stale ring. + news: the normal body (`active: false`) is returned instead of a stale ring. - `?deviceId=` selects the subscription (raw value, as sent to `joinAlert`); without it the caller's oldest subscription on the alert is used, mirroring the socket. The service hashes it before comparing. @@ -686,11 +686,21 @@ Response `200`: - Clearing is **ack-driven**, never response-driven: `alarmStatus` only *reports* debt (409 + frame) and clears nothing, so a 409 lost in transit does not forfeit the alarm — the next check-in finds the same row. +- **Receipt and dismissal are different facts.** A client acks a frame as soon as + it receives it (that is how an online device stops being told about the same + alarm on every poll). That writes `delivered_at`, which retires re-delivery but + is *not* a dismissal. Only `ackAlarm` writes `acked_at`, and only `acked_at` + counts towards the drain-to-empty rule — so the alert survives while phones are + still ringing even when everyone happened to be online. - Idempotent: a repeat ack (or an ack for an event the device never owed) returns - `acked: 0`, never an error. The WebSocket `ack` frame settles the same rows. -- `acked` counts the ledger rows cleared for this device. When the ack clears the - last unacked row of the alert, the alert drains and is destroyed (reason - `delivered`) — `destroyed: true` reports that intended outcome. Note that a + `acked: 0`, never an error. The WebSocket `ack` frame does NOT dismiss an + alarm — it only records the receipt. +- `acked` counts the ledger rows dismissed for this device. When the last + un-dismissed row of the alert is dismissed, the alert drains and is destroyed + (reason `delivered`) — `destroyed: true` reports that intended outcome. A + receipt (the WS `ack` frame) never counts here: it only sets `delivered_at`. + The drain check runs on every explicit dismissal, even when this call cleared + no row, because a receipted row is still undismissed. Note that a destroyed alert takes its subscriptions with it, so a device token for it stops resolving (the next call is `401 ERR_UNAUTHORIZED`, not `404`); clients must treat that as "the subscription is over", not as a retryable failure. diff --git a/database/migrations/011_delivery_receipts.sql b/database/migrations/011_delivery_receipts.sql new file mode 100644 index 0000000..c1cf40e --- /dev/null +++ b/database/migrations/011_delivery_receipts.sql @@ -0,0 +1,55 @@ +-- Copyright (C) 2026 Dcntrlized +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as published +-- by the Free Software Foundation, either version 3 of the License, or +-- (at your option) any later version. +-- +-- This program is distributed in the hope that it will be useful, +-- but WITHOUT ANY WARRANTY; without even the implied warranty of +-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +-- GNU Affero General Public License for more details. +-- +-- You should have received a copy of the GNU Affero General Public License +-- along with this program. If not, see . + +-- 011_delivery_receipts.sql — "received" is not "dismissed". +-- +-- The delivery ledger had one timestamp (`acked_at`) for two different facts, +-- and the clients ack a frame the moment they receive it (that is how a live +-- device stops being told about the same alarm on every poll). Because the +-- drain-to-empty rule destroys the alert when every row is acked, a single-device +-- alert was destroyed 27 ms after the owner triggered it — measured on an +-- Android 15 emulator: `triggered` then `destroyed/delivered` in the event log, +-- while the phone was still ringing. The owner then could not disarm +-- (`ERR_ALERT_NOT_FOUND`), the subscription went with the alert, and its device +-- token stopped resolving mid-alarm (401). +-- +-- So the ledger now records both facts: +-- * `delivered_at` — the device has the trigger (a live frame, a replay, or a +-- check-in). It stops re-delivery. It is NOT a dismissal. +-- * `acked_at` — the person holding the device dismissed it. Only this drives +-- the drain, so the alert ends when everyone has dismissed it, never because +-- everyone happened to be online. +-- +-- Additive and idempotent. Existing rows: a row with `acked_at` set keeps that +-- fact (it was a dismissal under the old contract too); a pending row simply has +-- no receipt yet, which is the correct reading of "we never recorded a delivery". + +\c dcntrlized + +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'alarm') THEN + RETURN; + END IF; + + ALTER TABLE alarm.alarm_deliveries ADD COLUMN IF NOT EXISTS delivered_at TIMESTAMPTZ; + + -- "Who still owes an alarm on this device" is the check-in query; the + -- partial index keeps it to the rows that can still answer yes. + DROP INDEX IF EXISTS alarm.alarm_deliveries_pending_idx; + CREATE INDEX IF NOT EXISTS alarm_deliveries_undelivered_idx + ON alarm.alarm_deliveries (alert_id, subscription_id) + WHERE acked_at IS NULL AND delivered_at IS NULL; +END $$; diff --git a/src/actions/ack.go b/src/actions/ack.go index 5001351..c83210b 100644 --- a/src/actions/ack.go +++ b/src/actions/ack.go @@ -125,13 +125,17 @@ func AckAlarm(w http.ResponseWriter, r *http.Request, payload any) { dbError(w, action, err) return } - destroyed, err := db.CompleteAlertDeliveries(r.Context(), alertID) - if err != nil { - dbError(w, action, err) - return - } - resp.Destroyed = destroyed } + // The drain check runs on every explicit dismissal, not only when this + // call cleared a row: the receipt ack marks a row delivered but not + // dismissed, so the person dismissing is what completes the set even + // when there was nothing left for THIS call to clear. + destroyed, err := db.CompleteAlertDeliveries(r.Context(), alertID) + if err != nil { + dbError(w, action, err) + return + } + resp.Destroyed = destroyed Debug(action, "POST", "alert=%s event=%s all=%t acked=%d destroyed=%t (device)", alertID, resp.EventID, eventID == uuid.Nil, resp.Acked, resp.Destroyed) writeJSON(w, http.StatusOK, resp) @@ -163,13 +167,13 @@ func AckAlarm(w http.ResponseWriter, r *http.Request, payload any) { dbError(w, action, err) return } - destroyed, err := db.CompleteAlertDeliveries(r.Context(), alertID) - if err != nil { - dbError(w, action, err) - return - } - resp.Destroyed = destroyed } + destroyed, err := db.CompleteAlertDeliveries(r.Context(), alertID) + if err != nil { + dbError(w, action, err) + return + } + resp.Destroyed = destroyed } Debug(action, "POST", "alert=%s event=%s all=%t acked=%d destroyed=%t", diff --git a/src/db/db.go b/src/db/db.go index 23bb1b3..cedf92a 100644 --- a/src/db/db.go +++ b/src/db/db.go @@ -112,6 +112,12 @@ func Close() { if instance != nil && instance.Pool != nil { instance.Pool.Close() } + // Reset the singleton so a later NewDB()/Pool() opens a fresh pool instead + // of handing out the closed one. The probes in this package each open and + // close their own pool, and a sequential run of two probes must not inherit + // the previous one's closed pool. + instance = nil + once = sync.Once{} } // redact strips the password from a DSN before it can reach a log line: diff --git a/src/db/deliveries.go b/src/db/deliveries.go index 64842cb..b24b1a4 100644 --- a/src/db/deliveries.go +++ b/src/db/deliveries.go @@ -59,6 +59,7 @@ func PendingTrigger(ctx context.Context, alertID, userID uuid.UUID, deviceID str WHERE d.alert_id = $1 AND d.subscription_id = (SELECT id FROM target) AND d.acked_at IS NULL + AND d.delivered_at IS NULL ORDER BY e.created_at DESC, e.id DESC LIMIT 1` @@ -94,6 +95,7 @@ func PendingReplay(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]Pe WHERE d.alert_id = $1 AND d.subscription_id = $2 AND d.acked_at IS NULL + AND d.delivered_at IS NULL AND NOT EXISTS ( SELECT 1 FROM alarm.alarm_events x WHERE x.alert_id = d.alert_id @@ -129,7 +131,7 @@ func PendingReplay(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]Pe func AckDelivery(ctx context.Context, alertID, subscriptionID, eventID uuid.UUID) (int64, error) { tag, err := Pool().Exec(ctx, `UPDATE alarm.alarm_deliveries - SET acked_at = now() + SET acked_at = now(), delivered_at = coalesce(delivered_at, now()) WHERE alert_id = $1 AND subscription_id = $2 AND event_id = $3 AND acked_at IS NULL`, alertID, subscriptionID, eventID) @@ -150,7 +152,7 @@ func AckDelivery(ctx context.Context, alertID, subscriptionID, eventID uuid.UUID func AckAlarmDelivery(ctx context.Context, alertID, userID uuid.UUID, deviceID string, eventID uuid.UUID) (int64, error) { const q = ` UPDATE alarm.alarm_deliveries d - SET acked_at = now() + SET acked_at = now(), delivered_at = coalesce(d.delivered_at, now()) WHERE d.alert_id = $1 AND d.event_id = $2 AND d.acked_at IS NULL @@ -179,7 +181,8 @@ func HasPendingDelivery(ctx context.Context, alertID, subscriptionID uuid.UUID) err := Pool().QueryRow(ctx, `SELECT EXISTS ( SELECT 1 FROM alarm.alarm_deliveries - WHERE alert_id = $1 AND subscription_id = $2 AND acked_at IS NULL)`, + WHERE alert_id = $1 AND subscription_id = $2 + AND acked_at IS NULL AND delivered_at IS NULL)`, alertID, subscriptionID).Scan(&owed) if err != nil { return false, err @@ -199,6 +202,7 @@ func HasPendingDeliveryForUser(ctx context.Context, alertID, userID uuid.UUID, d JOIN alarm.subscriptions s ON s.id = d.subscription_id WHERE d.alert_id = $1 AND d.acked_at IS NULL + AND d.delivered_at IS NULL AND s.subscriber_user_id = $2 AND ($3 = '' OR s.device_id = $3))`, alertID, userID, deviceHash).Scan(&owed) @@ -217,7 +221,7 @@ func HasPendingDeliveryForUser(ctx context.Context, alertID, userID uuid.UUID, d func AckAllDeliveries(ctx context.Context, alertID, subscriptionID uuid.UUID) (int64, error) { tag, err := Pool().Exec(ctx, `UPDATE alarm.alarm_deliveries - SET acked_at = now() + SET acked_at = now(), delivered_at = coalesce(delivered_at, now()) WHERE alert_id = $1 AND subscription_id = $2 AND acked_at IS NULL`, alertID, subscriptionID) if err != nil { @@ -231,7 +235,7 @@ func AckAllDeliveries(ctx context.Context, alertID, subscriptionID uuid.UUID) (i func AckAllForSubscriber(ctx context.Context, alertID, userID uuid.UUID, deviceHash string) (int64, error) { tag, err := Pool().Exec(ctx, `UPDATE alarm.alarm_deliveries d - SET acked_at = now() + SET acked_at = now(), delivered_at = coalesce(d.delivered_at, now()) WHERE d.alert_id = $1 AND d.acked_at IS NULL AND d.subscription_id IN ( @@ -245,3 +249,48 @@ func AckAllForSubscriber(ctx context.Context, alertID, userID uuid.UUID, deviceH } return tag.RowsAffected(), nil } + +// MarkDelivered records that the device has the trigger, without treating that +// as a dismissal. +// +// Clients ack a frame the moment they receive it so the service stops offering +// the same alarm on every poll. That is a receipt, not a decision by the person +// holding the phone: the reveal screen may still be behind a locked screen, and +// the alarm is still ringing. Recording it separately is what stops the drain +// rule from destroying an alert while it is still sounding. +// +// Idempotent and best-effort by construction: a repeat, a malformed event id or +// an admin connection match nothing. +func MarkDelivered(ctx context.Context, alertID, subscriptionID, eventID uuid.UUID) (int64, error) { + tag, err := Pool().Exec(ctx, + `UPDATE alarm.alarm_deliveries + SET delivered_at = now() + WHERE alert_id = $1 AND subscription_id = $2 AND event_id = $3 + AND delivered_at IS NULL AND acked_at IS NULL`, + alertID, subscriptionID, eventID) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + +// MarkAllDeliveredForUser is MarkDelivered for the account path: everything this +// caller's device still has not received is now received. +func MarkAllDeliveredForUser(ctx context.Context, alertID, userID uuid.UUID, deviceHash string) (int64, error) { + tag, err := Pool().Exec(ctx, + `UPDATE alarm.alarm_deliveries d + SET delivered_at = now() + WHERE d.alert_id = $1 + AND d.acked_at IS NULL + AND d.delivered_at IS NULL + AND d.subscription_id IN ( + SELECT s.id FROM alarm.subscriptions s + WHERE s.alert_id = $1 + AND s.subscriber_user_id = $2 + AND ($3 = '' OR s.device_id = $3))`, + alertID, userID, deviceHash) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} diff --git a/src/db/deliveries_db_test.go b/src/db/deliveries_db_test.go index 1c77734..2f1a44e 100644 --- a/src/db/deliveries_db_test.go +++ b/src/db/deliveries_db_test.go @@ -77,9 +77,13 @@ func TestLedgerProbe(t *testing.T) { } return a } + // Each subscription needs its own device id: migration 010 enforces + // (alert_id, device_id) uniqueness so one device has at most one + // subscription per alert. join := func(alertID, user uuid.UUID) *Subscription { s, err := CreateSubscription(ctx, &Subscription{ - AlertID: alertID, SubscriberUserID: user, DeviceID: "sha256:" + strings.Repeat("a", 64), + AlertID: alertID, SubscriberUserID: user, + DeviceID: "sha256:" + strings.ReplaceAll(user.String(), "-", ""), }) if err != nil { t.Fatalf("CreateSubscription: %v", err) @@ -102,8 +106,10 @@ func TestLedgerProbe(t *testing.T) { join(drain.ID, sub1) join(drain.ID, sub2) join(drain.ID, sub3) // third enabled subscriber: acking a different device's row must not count + disabledUser := uuid.New() disabled, err := CreateSubscription(ctx, &Subscription{ - AlertID: drain.ID, SubscriberUserID: uuid.New(), DeviceID: "sha256:" + strings.Repeat("b", 64), + AlertID: drain.ID, SubscriberUserID: disabledUser, + DeviceID: "sha256:" + strings.ReplaceAll(disabledUser.String(), "-", ""), }) if err != nil { t.Fatalf("disabled subscription: %v", err) diff --git a/src/db/device_ledger_db_test.go b/src/db/device_ledger_db_test.go index a60b5f1..11798de 100644 --- a/src/db/device_ledger_db_test.go +++ b/src/db/device_ledger_db_test.go @@ -92,35 +92,43 @@ func TestDeviceLedgerProbe(t *testing.T) { t.Fatalf("account owes after trigger: %v %v", ok, err) } - // The device dismissal that names no event clears its whole set; a repeat - // (or a named ack afterwards) is an idempotent no-op. - cleared, err := AckAllDeliveries(ctx, a.ID, device.ID) - if err != nil || cleared != 1 { - t.Fatalf("AckAllDeliveries = %d, %v; want 1", cleared, err) - } - if again, err := AckDelivery(ctx, a.ID, device.ID, event.ID); err != nil || again != 0 { - t.Fatalf("AckDelivery after ack-all = %d, %v; want 0", again, err) + // A receipt is not a dismissal. The clients ack a frame the moment they get + // it, and that used to be written to acked_at — which destroyed the alert + // while it was still ringing (measured on the emulator: 27 ms after the + // trigger). A receipt settles re-delivery only. + marked, err := MarkDelivered(ctx, a.ID, device.ID, event.ID) + if err != nil || marked != 1 { + t.Fatalf("MarkDelivered = %d, %v; want 1", marked, err) } if ok, err := HasPendingDelivery(ctx, a.ID, device.ID); err != nil || ok { - t.Fatalf("device still owes after ack-all: %v %v", ok, err) + t.Fatalf("device still owes after a receipt: %v %v", ok, err) } - replay, err := PendingReplay(ctx, a.ID, device.ID) - if err != nil || len(replay) != 0 { - t.Fatalf("device replay after ack-all = %d rows, %v; want 0", len(replay), err) + if replay, err := PendingReplay(ctx, a.ID, device.ID); err != nil || len(replay) != 0 { + t.Fatalf("device replay after a receipt = %d rows, %v; want 0", len(replay), err) + } + if destroyed, err := CompleteAlertDeliveries(ctx, a.ID); err != nil || destroyed { + t.Fatalf("a receipt destroyed the alert: %v %v", destroyed, err) } - // The other subscriber's debt is untouched: acking is per subscription. + // The dismissal is what the drain counts. A device that received and later + // dismissed must not end the alert while another subscriber has not. + if got, err := AckDelivery(ctx, a.ID, device.ID, event.ID); err != nil || got != 1 { + t.Fatalf("AckDelivery after a receipt = %d, %v; want 1 (the dismissal still has to be recorded)", got, err) + } if destroyed, err := CompleteAlertDeliveries(ctx, a.ID); err != nil || destroyed { - t.Fatalf("alert destroyed while another subscriber still owes: %v %v", destroyed, err) + t.Fatalf("one dismissal destroyed an alert another subscriber never dismissed: %v %v", destroyed, err) } if ok, err := HasPendingDeliveryForUser(ctx, a.ID, account, ""); err != nil || !ok { - t.Fatalf("account debt lost when the device acked: %v %v", ok, err) + t.Fatalf("account debt lost when the device dismissed: %v %v", ok, err) + } + if got, err := MarkAllDeliveredForUser(ctx, a.ID, account, ""); err != nil || got != 1 { + t.Fatalf("MarkAllDeliveredForUser = %d, %v; want 1", got, err) } if got, err := AckAllForSubscriber(ctx, a.ID, account, ""); err != nil || got != 1 { t.Fatalf("AckAllForSubscriber = %d, %v; want 1", got, err) } if destroyed, err := CompleteAlertDeliveries(ctx, a.ID); err != nil || !destroyed { - t.Fatalf("last ack did not destroy the alert: %v %v", destroyed, err) + t.Fatalf("the last dismissal did not destroy the alert: %v %v", destroyed, err) } // The alert, its subscriptions and the device's token row are all gone. diff --git a/src/realtime/client.go b/src/realtime/client.go index f2c5da5..c82fbc2 100644 --- a/src/realtime/client.go +++ b/src/realtime/client.go @@ -124,11 +124,21 @@ func (c *client) readPump() { } } -// ackDelivery marks the ledger row for a client ack. It is best-effort and -// idempotent: a malformed event id, an admin connection (no subscription) or a -// repeated ack are all silent no-ops. A database failure is logged, never fatal — -// the client already has the alarm, and the worst case is one duplicate offline -// signal, not a dropped ring. +// ackDelivery records a *receipt*: the client has the frame, so the service must +// not offer it again on the next check-in or reconnect. +// +// It deliberately does not dismiss the alarm. Clients ack automatically the +// moment a frame arrives — the person may still be asleep behind a lock screen +// and the siren is still sounding — so treating that ack as "dismissed" let the +// drain-to-empty rule destroy the alert 27 ms after the owner triggered it +// (measured on an Android 15 emulator: the phone was ringing for an alert that +// no longer existed). Dismissal is the explicit `ackAlarm` action; only that +// drives the drain. +// +// Best-effort and idempotent: a malformed event id, an admin connection (no +// subscription) or a repeated receipt are all silent no-ops. A database failure +// is logged, never fatal — the client already has the alarm, and the worst case +// is one duplicate check-in answer, not a dropped ring. func (c *client) ackDelivery(eventID string) { if c.identity.SubscriptionID == nil { return @@ -139,14 +149,14 @@ func (c *client) ackDelivery(eventID string) { } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() - cleared, err := db.AckDelivery(ctx, c.identity.AlertID, *c.identity.SubscriptionID, id) + marked, err := db.MarkDelivered(ctx, c.identity.AlertID, *c.identity.SubscriptionID, id) if err != nil { - c.hub.opts.Logf("ws: ledger ack failed alert=%s event=%s: %v", c.identity.AlertID, id, err) + c.hub.opts.Logf("ws: delivery receipt failed alert=%s event=%s: %v", c.identity.AlertID, id, err) return } - if cleared == 0 { - // A repeat ack (or one for an event this device never owed) is a no-op: - // nothing to record and no drain to check. + if marked == 0 { + // A repeated receipt (or one for an event this device was never offered) + // is a no-op: nothing to record. return } // Anonymous subscribers have no account to record analytics against: the @@ -158,9 +168,6 @@ func (c *client) ackDelivery(eventID string) { } else if err := db.RecordDelivered(ctx, c.identity.AlertID, c.identity.UserID); err != nil { c.hub.opts.Logf("ws: delivered analytics alert=%s: %v", c.identity.AlertID, err) } - if _, err := db.CompleteAlertDeliveries(ctx, c.identity.AlertID); err != nil { - c.hub.opts.Logf("ws: drain check alert=%s: %v", c.identity.AlertID, err) - } } // writePump owns all writes except the WriteControl calls in readPump (which