diff --git a/server/resources/migrations/124_add_custodian.down.sql b/server/resources/migrations/124_add_custodian.down.sql new file mode 100644 index 0000000000..833207fb76 --- /dev/null +++ b/server/resources/migrations/124_add_custodian.down.sql @@ -0,0 +1,15 @@ +drop table custodian; + +-- Recreate the trigger that cleans up triples_size_updates on attr delete. +create or replace function clean_triples_size_updates() +returns trigger as $$ +begin + delete from triples_size_updates where triples_size_updates.attr_id = old.id; + return old; +end; +$$ language plpgsql; + +create trigger clean_triples_size_updates_trigger +before delete on attrs +for each row +execute function clean_triples_size_updates(); diff --git a/server/resources/migrations/124_add_custodian.up.sql b/server/resources/migrations/124_add_custodian.up.sql new file mode 100644 index 0000000000..6b05944bd7 --- /dev/null +++ b/server/resources/migrations/124_add_custodian.up.sql @@ -0,0 +1,52 @@ +create table custodian ( + id uuid primary key default gen_random_uuid(), + -- The app being deleted. Cascade so the terminal `app` delete cleans up its + -- own plan rows for free. + app_id uuid not null references apps(id) on delete cascade, + -- Set when a unit of work is scoped to a single attr (e.g. deleting one + -- attr's triples). Null means the whole app. + attr_id uuid references attrs(id) on delete cascade, + -- What this row deletes: 'triples' | 'transactions' | 'attrs' | 'attr' | 'app' + type text not null, + -- The step this one depends on: it can't run until that step is done. Forms a + -- chain, e.g. for an app: app depends on transactions depends on triples. A + -- step finishes by deleting its row; `on delete set null` then clears this + -- pointer on the dependent, so the runnable row is simply the one with + -- depends_on is null. + depends_on uuid references custodian(id) on delete set null, + -- The worker that owns this row (null when unclaimed), set on claim. Doubles + -- as an owner tag. A worker heartbeats by bumping updated_at as it works; the + -- reaper frees a row (clears worker_id) whose updated_at has gone stale. + worker_id text, + -- 'waiting' (runnable) -> 'working' (claimed by a worker) and back to 'waiting' + -- on a failed attempt; set to 'failed' once processing has errored enough times + -- (see `attempts`) so it stops being retried and can be investigated. + status text not null default 'waiting', + -- How many times processing this row has errored. We retry a few times before + -- giving up and marking it 'failed', since some failures are transient. + attempts integer not null default 0, + -- The error message from the most recent failure. + error text, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + -- At most one row per (app, type, attr). `nulls not distinct` so two whole-app + -- rows (attr_id is null) of the same type collide instead of duplicating. + constraint custodian_unique unique nulls not distinct (app_id, type, attr_id) +); + +create index custodian_attr_id on custodian (attr_id); +create index custodian_depends_on on custodian (depends_on); + +create index custodian_claimable on custodian (created_at) + where depends_on is null and worker_id is null and status = 'waiting'; + +create trigger update_custodian_updated_at +before update on custodian +for each row +execute function update_updated_at_column(); + +-- Drop the trigger that cleaned up triples_size_updates on attr delete. The +-- updater clears out the rows for us, so there's no need to delete them here. +-- When we delete an app, it causes us to delete too much. +drop trigger if exists clean_triples_size_updates_trigger on attrs; +drop function if exists clean_triples_size_updates(); diff --git a/server/src/instant/core.clj b/server/src/instant/core.clj index 500af5ebb1..1f99a02c6b 100644 --- a/server/src/instant/core.clj +++ b/server/src/instant/core.clj @@ -65,6 +65,7 @@ [instant.util.posthog :as posthog] [instant.util.tracer :as tracer] [instant.hard-deletion-sweeper :as hard-deletion-sweeper] + [instant.custodian :as custodian] [instant.webhook-routes :as webhook-routes] [instant.webhook-processor :as webhook-processor] [ring.middleware.cookies :refer [CookieDateTime]] @@ -359,6 +360,12 @@ (future (tracer/with-span! {:name "stop-join-room-logger"} (join-room-logger/stop))) + (future + (tracer/with-span! {:name "stop-hard-deletion-sweeper"} + (hard-deletion-sweeper/stop))) + (future + (tracer/with-span! {:name "stop-custodian"} + (custodian/stop))) (future (when (posthog/enabled?) (tracer/with-span! {:name "stop-posthog"} @@ -478,6 +485,8 @@ (storage-sweeper/start)) (with-log-init :hard-deletion-sweeper (hard-deletion-sweeper/start)) + (with-log-init :custodian + (custodian/start)) (with-log-init :rate-limit-sweeper (rate-limit/start)) (with-log-init :wal-log-truncator diff --git a/server/src/instant/custodian.clj b/server/src/instant/custodian.clj new file mode 100644 index 0000000000..7609fb2879 --- /dev/null +++ b/server/src/instant/custodian.clj @@ -0,0 +1,568 @@ +(ns instant.custodian + "Deletes an app (or a single attr) in small, bounded, committed batches so that + we don't hand the invalidator one giant WAL transaction. + + The plan for a deletion is stored as a chain of `custodian` rows linked by + `depends_on`: + + app: triples <- transactions <- attrs <- app + attr: triples <- attr + + `depends_on` points at the step that must finish first, so a row is runnable + once it no longer depends on any previous step. Each worker claims one runnable + row, drains it in `batch-size` chunks (each chunk its own committed transaction), + then deletes it, makes its dependent runnable. The terminal `app`/`attr` step + deletes the app/attr itself. + + One worker runs per machine. It loops while there is runnable work and pauses + for `:custodian-idle-sleep-ms` when there isn't." + (:require + [chime.core :as chime-core] + [clojure.string] + [instant.config :as config] + [instant.db.model.attr :as attr-model] + [instant.db.transaction :as tx] + [instant.flags :as flags] + [instant.jdbc.aurora :as aurora] + [instant.jdbc.sql :as sql] + [instant.util.async :as ua] + [instant.util.exception :as ex] + [instant.util.hsql :as uhsql] + [instant.util.tracer :as tracer]) + (:import + (clojure.lang ExceptionInfo) + (java.io PrintWriter StringWriter) + (java.lang AutoCloseable) + (java.time Duration Instant))) + +;; ---------- +;; Config + +(defn- batch-size [] (flags/flag :custodian-batch-size 1000)) +(defn- idle-sleep-ms [] (flags/flag :custodian-idle-sleep-ms 5000)) + +;; Attrs are deleted one at a time: each attr delete cascades to that attr's +;; triples, and right after the triple sweep those are dead-but-not-yet-vacuumed +;; index entries, so the cascade scan is slow. One attr per statement keeps each +;; cascade bounded (and lets autovacuum catch up between batches). +(defn- attr-batch-size [] (flags/flag :custodian-attr-batch-size 1)) + +;; Lease long enough that a batch (plus its lease renewal) comfortably finishes, +;; short enough that a dead worker's row gets picked up again reasonably soon. +(defn- lease-timeout-seconds [] (flags/flag :custodian-lease-timeout-seconds 120)) + +;; How many times processing a row may error before we give up and mark it +;; 'failed'. Retries handle transient failures (a lost connection, a deadlock). +(defn- max-attempts [] (flags/flag :custodian-max-attempts 3)) + +;; How long a 'failed' row lingers before the reaper deletes it (and its stuck +;; dependents): long enough to notice and investigate, not forever. +(defn- failed-retention-seconds [] (flags/flag :custodian-failed-retention-seconds 86400)) + +;; ---------- +;; Enqueue + +;; Relies on the (app_id, type, attr_id) unique constraint: re-enqueuing an +;; in-flight plan throws and rolls the whole statement back (no partial chain). +;; The enqueue fns catch that violation, making a re-enqueue an idempotent no-op. +(def enqueue-app-deletion-q + (uhsql/preformat + {:with [[:triples-row {:insert-into :custodian + :values [{:app-id :?app-id :type [:inline "triples"]}] + :returning :id}] + [:tx-row {:insert-into :custodian + :values [{:app-id :?app-id + :type [:inline "transactions"] + :depends-on {:select :id :from :triples-row}}] + :returning :id}] + [:attrs-row {:insert-into :custodian + :values [{:app-id :?app-id + :type [:inline "attrs"] + :depends-on {:select :id :from :tx-row}}] + :returning :id}]] + :insert-into :custodian + :values [{:app-id :?app-id + :type [:inline "app"] + :depends-on {:select :id :from :attrs-row}}]})) + +(def enqueue-attr-deletion-q + (uhsql/preformat + {:with [[:triples-row {:insert-into :custodian + :values [{:app-id :?app-id + :attr-id :?attr-id + :type [:inline "triples"]}] + :returning :id}]] + :insert-into :custodian + :values [{:app-id :?app-id + :attr-id :?attr-id + :type [:inline "attr"] + :depends-on {:select :id :from :triples-row}}]})) + +(defn enqueue-app-deletion! + "Idempotently enqueues the chain triples <- transactions <- app for an app, + atomically so a worker never sees a partial chain." + ([params] (enqueue-app-deletion! (aurora/conn-pool :write) params)) + ([conn {:keys [app-id]}] + (try + (sql/do-execute! ::enqueue-app-deletion! + conn + (uhsql/formatp enqueue-app-deletion-q {:app-id app-id})) + (catch ExceptionInfo e + ;; Plan already in flight; the unique constraint rolled the insert back. + (when-not (= ::ex/record-not-unique (::ex/type (ex-data e))) + (throw e)))))) + +(defn enqueue-attr-deletion! + "Idempotently enqueues the chain triples <- attr for a single attr." + ([params] (enqueue-attr-deletion! (aurora/conn-pool :write) params)) + ([conn {:keys [app-id attr-id]}] + (try + (sql/do-execute! ::enqueue-attr-deletion! + conn + (uhsql/formatp enqueue-attr-deletion-q {:app-id app-id + :attr-id attr-id})) + (catch ExceptionInfo e + ;; Plan already in flight; the unique constraint rolled the insert back. + (when-not (= ::ex/record-not-unique (::ex/type (ex-data e))) + (throw e)))))) + +;; ---------- +;; Claim / lease + +(def claim-row-q + (uhsql/preformat + {:with [[:claimed {:select :id + :from :custodian + :where [:and + [:= :status [:inline "waiting"]] + [:is :depends-on nil] + [:is :worker-id nil]] + :order-by [:created-at] + :for [:update :skip-locked] + :limit [:inline 1]}]] + :update :custodian + :set {:worker-id :?worker-id :status [:inline "working"]} + :from :claimed + :where [:= :custodian.id :claimed.id] + :returning :custodian.*})) + +(def finish-q + (uhsql/preformat {:delete-from :custodian + :where [:and [:= :id :?id] [:= :worker-id :?worker-id]]})) + +(def release-q + (uhsql/preformat {:update :custodian + :set {:worker-id nil :status [:inline "waiting"]} + :where [:and [:= :id :?id] [:= :worker-id :?worker-id]]})) + +;; Record a failed attempt: bump `attempts` and stash the error. Below +;; `max-attempts` we hand the row back (status 'waiting', worker_id cleared) so +;; it gets retried; once we hit the limit we flip it to 'failed' so it stops +;; being retried and can be investigated. +(def fail-q + (uhsql/preformat {:update :custodian + :set {:attempts [:+ :attempts :1] + :error :?error + :worker-id [:case [:>= [:+ :attempts :1] :?max-attempts] + :worker-id :else nil] + :status [:case [:>= [:+ :attempts :1] :?max-attempts] + [:inline "failed"] :else [:inline "waiting"]]} + :where [:and [:= :id :?id] [:= :worker-id :?worker-id]]})) + +(def reap-stuck-q + (uhsql/preformat {:update :custodian + :set {:worker-id nil :status [:inline "waiting"]} + :where [:and [:= :status [:inline "working"]] + [:is-not :worker-id nil] [:< :updated-at :?stale]] + :returning :id})) + +;; Delete rows that have been 'failed' longer than the retention window, plus +;; everything that (transitively) depends on them. We take the dependents too +;; because `depends_on`'s `on delete set null` would otherwise make a failed +;; step's dependent runnable, letting the plan skip a step that never finished. +(def reap-failed-q + (uhsql/preformat + {:with-recursive [[:plan {:union [{:select :id + :from :custodian + :where [:and [:= :status [:inline "failed"]] + [:< :updated-at :?stale]]} + {:select :c.id + :from [[:custodian :c]] + :join [[:plan :p] [:= :c.depends-on :p.id]]}]}]] + :delete-from :custodian + :where [:in :id {:select :id :from :plan}] + :returning :id})) + +(defn claim-row! + "Claims one runnable, unowned row (depends_on IS NULL, worker_id IS NULL) under + FOR UPDATE SKIP LOCKED so workers on different machines cooperate. Sets + worker_id to this machine. Returns the claimed row, or nil when there's no + runnable work." + [conn] + (sql/execute-one! ::claim-row! + conn + (uhsql/formatp claim-row-q {:worker-id @config/process-id}))) + +(defn- finish! [conn id] + ;; Deletes the finished step; the FK's `on delete set null` clears its + ;; dependent's depends_on, making that dependent runnable. Ownership-guarded so + ;; a worker that lost the row can't remove it. For the terminal app/attr step + ;; the row is already gone via cascade, so this is a no-op. + (sql/do-execute! ::finish! conn (uhsql/formatp finish-q {:id id + :worker-id @config/process-id}))) + +(defn- release! [conn id] + ;; Give the row back (clear our worker_id) so another worker can reclaim it + ;; immediately instead of waiting for the reaper. Ownership-guarded. + (sql/do-execute! ::release! conn (uhsql/formatp release-q {:id id + :worker-id @config/process-id}))) + +(defn- fail! [conn id error] + ;; Record a failed attempt: retry (hand the row back) until we've hit + ;; `max-attempts`, then mark it failed so it stops being retried and can be + ;; investigated. Ownership-guarded, so a worker that lost the row can't fail it. + (sql/do-execute! ::fail! conn (uhsql/formatp fail-q {:id id + :error error + :max-attempts (max-attempts) + :worker-id @config/process-id}))) + +(defn reap-stuck! + "Frees 'working' rows whose owner stopped heartbeating (updated_at older than + the lease timeout), setting them back to 'waiting' so another worker can + reclaim them. Returns the freed row ids." + ([] (reap-stuck! (aurora/conn-pool :write))) + ([conn] + (let [stale (.minusSeconds (Instant/now) (lease-timeout-seconds))] + (sql/execute! ::reap-stuck! conn (uhsql/formatp reap-stuck-q {:stale stale}))))) + +(defn reap-failed! + "Deletes plans that have been 'failed' longer than the retention window (the + failed row plus its stuck dependents). The app stays marked for deletion, so + the sweeper re-enqueues a fresh plan later. Returns the deleted row ids." + ([] (reap-failed! (aurora/conn-pool :write))) + ([conn] + (let [stale (.minusSeconds (Instant/now) (failed-retention-seconds))] + (sql/execute! ::reap-failed! conn (uhsql/formatp reap-failed-q {:stale stale}))))) + +;; ---------- +;; Backpressure + +;; Our deletes produce WAL the invalidator and other logical consumers must keep +;; up with. A chime samples the worst active replication lag every minute; while +;; it's over the threshold the worker backs off so we don't outrun them. +(defn max-replication-lag-bytes [conn] + (:lag (sql/select-one + ::max-replication-lag-bytes + conn + ["select max(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) as lag + from pg_replication_slots + where confirmed_flush_lsn is not null"]))) + +(defn- nudge-slots-in-dev! [conn] + ;; When the db is idle the invalidator and aggregator replication slots never + ;; flush and the lag climbs until the worker parks. The aggregator only decodes + ;; `triples`, so we need a real triples change: add then delete a flag on the + ;; config app — it leaves nothing behind but produces the triples WAL (plus the + ;; transactions anchor) that advances both slots. + (when (clojure.string/includes? config/server-origin "localhost") + (let [attrs (attr-model/get-by-app-id conn config/instant-config-app-id) + id-attr (attr-model/resolve-attr-id attrs "flags" "id") + setting-attr (attr-model/resolve-attr-id attrs "flags" "setting") + value-attr (attr-model/resolve-attr-id attrs "flags" "value") + entity-id (random-uuid)] + (tx/transact! conn attrs config/instant-config-app-id + [[:add-triple entity-id id-attr entity-id] + [:add-triple entity-id setting-attr (str ::bump-slots)] + [:add-triple entity-id value-attr "true"]]) + (tx/transact! conn attrs config/instant-config-app-id + [[:delete-entity entity-id]])))) + +(defn sample-lag! [backing-off?] + (try + (let [threshold (flags/flag :custodian-max-replication-lag-bytes (* 100 1024 1024)) + lag (or (max-replication-lag-bytes (aurora/conn-pool :read)) 0) + paused? (> lag threshold) + [was-paused?] (reset-vals! backing-off? paused?)] + ;; Nudge once we're halfway to the limit so the slot keeps advancing and we + ;; ideally never actually cross it (in dev this is usually just an idle slot). + (when (> lag (quot threshold 2)) + (try + (nudge-slots-in-dev! (aurora/conn-pool :write)) + (catch Throwable _ nil))) + ;; Log the transition (once per pause/resume, not every sample). + (when (not= was-paused? paused?) + (tracer/record-info! {:name (if paused? "custodian/pause" "custodian/resume") + :attributes {:lag-bytes lag + :threshold-bytes threshold}}))) + (catch Throwable t + (tracer/record-exception-span! t {:name "custodian/sample-lag-error"})))) + +(defn- should-pause? + "Whether the worker should hold off on writes right now: replication is lagging + (`backing-off?`), we're failing over and must leave the old primary alone, or + the custodian is disabled. A pause is not a quit — it's released once the + condition clears; only `@stop?` exits the worker loop." + [backing-off?] + (or @backing-off? + (flags/failing-over?) + (flags/custodian-disabled?))) + +(defn- await-capacity! + "Blocks while `should-pause?` (replication lag or failover) so we don't outrun + the WAL consumers or write to a primary we're failing away from. Returns early + if the worker is stopping. A paused drain resumes where it left off once the + condition clears, without ever exiting the worker loop." + [stop? backing-off?] + (while (and (should-pause? backing-off?) (not @stop?)) + (Thread/sleep (long (idle-sleep-ms))))) + +;; ---------- +;; Processing + +;; Message the batch guard raises when the app/attr is no longer marked for +;; deletion. Shared so the query and the error check can't drift. +(def not-deleted-message "not marked for deletion") + +;; Each batch, in one statement: heartbeat the custodian row (bump updated_at) +;; while asserting we still own it, check the app/attr is still marked for +;; deletion, delete up to :limit rows, and report the count. If we lost ownership, +;; the row is gone, or the app/attr is no longer marked, it raises and unwinds the +;; drain. The marked check is its own CTE (referenced once in the final select), +;; so it doesn't slow the per-row scan that finds rows to delete. +(defn- delete-batch-query [table attr-scoped?] + (let [marked-for-deletion + [:case (if attr-scoped? + [:exists {:select :id :from :attrs + :where [:and [:= :id :?attr-id] [:is-not :deletion-marked-at nil]]}] + [:exists {:select :id :from :apps + :where [:and [:= :id :?app-id] [:is-not :deletion-marked-at nil]]}]) + true + :else [:raise_exception_message [:inline not-deleted-message]]]] + (uhsql/preformat + {:with [[:heartbeat {:update :custodian + :set {:updated-at :%now} + :where [:and + [:= :id :?id] + [:case [:= :worker-id :?worker-id] true + :else [:raise_exception_message + [:inline "custodian row not owned by this worker"]]]] + :returning :id}] + [:marked {:select [[marked-for-deletion :ok]]}] + [:to-delete {:select :ctid + :from table + :where (if attr-scoped? + [:and [:= :app-id :?app-id] [:= :attr-id :?attr-id]] + [:= :app-id :?app-id]) + :limit :?limit + :for :update}] + [:deleted {:delete-from table + :where [:in :ctid {:select :ctid :from :to-delete}] + :returning :*}]] + :select [[{:select :%count.* :from :deleted} :deleted] + [[:case [:exists {:select :id :from :heartbeat}] true + :else [:raise_exception_message + [:inline "custodian row no longer exists"]]] + :present] + [{:select :ok :from :marked} :marked]]}))) + +(def delete-app-triples-q (delete-batch-query :triples false)) +(def delete-attr-triples-q (delete-batch-query :triples true)) +(def delete-app-transactions-q (delete-batch-query :transactions false)) +(def delete-app-attrs-q (delete-batch-query :attrs false)) + +(defn- drain! + "Deletes rows via `q` in bounded, committed batches of `limit` (default + `batch-size`), heartbeating and asserting ownership per batch (see + `delete-batch-query`). `tag` labels the batch query (e.g. ::drain-app-attrs). + Returns ::completed when the table is drained, or ::stopped if the worker was + asked to stop mid-drain. Losing ownership throws and unwinds the drain." + ([tag stop? backing-off? conn id q params] + (drain! tag stop? backing-off? conn id q params (batch-size))) + ([tag stop? backing-off? conn id q params limit] + (loop [] + (await-capacity! stop? backing-off?) + (if @stop? + ::stopped + (let [deleted (-> (sql/do-execute! tag + conn + (uhsql/formatp q (assoc params + :id id + :worker-id @config/process-id + :limit limit))) + first + :deleted)] + (if (and deleted (pos? deleted)) + (recur) + ::completed)))))) + +(defn- delete-app! + "Deletes the app at the final stage, but checks if it is still marked as deleted first." + [conn app-id] + (sql/do-execute! ::delete-app! + conn + ["delete from apps where id = ? and deletion_marked_at is not null" app-id])) + +(defn- delete-attr! + "Deletes the attr at the final stage, but checks if it is still marked as deleted first." + [conn app-id attr-id] + (sql/do-execute! ::delete-app! + conn + ["delete from attrs where app_id = ? and id = ? and deletion_marked_at is not null" app-id attr-id])) + +(defn- not-deleted-error? + "True if `e` is the batch guard's raise for an app/attr that isn't marked for + deletion (vs. some other failure like losing ownership)." + [e] + (= not-deleted-message (-> e ex-data ::ex/pg-error-data :server-message))) + +(defn- abort-plan! + "The app/attr has a deletion plan but isn't marked for deletion (a bug, or a + restore that raced the sweeper). Log an error and delete the plan instead of + deleting live data." + [conn app-id attr-id e] + (tracer/with-new-trace-root + (tracer/record-exception-span! e {:name "custodian/plan-not-marked-for-deletion" + :attributes {:app-id app-id :attr-id attr-id}})) + (if attr-id + (sql/do-execute! ::abort-attr-plan conn + ["delete from custodian where app_id = ?::uuid and attr_id = ?::uuid" app-id attr-id]) + (sql/do-execute! ::abort-app-plan conn + ["delete from custodian where app_id = ?::uuid and attr_id is null" app-id]))) + +(defn- error-string + "The exception's message plus its full stack trace (including causes), for the + custodian row's `error` column so a failure has enough context to investigate." + [^Throwable e] + (let [sw (StringWriter.)] + (.printStackTrace e (PrintWriter. sw)) + (str sw))) + +(defn- process-row! [stop? backing-off? conn {:keys [id type app_id attr_id]}] + (tracer/with-span! {:name "custodian/process-row" + :attributes {:id id :type type :app-id app_id :attr-id attr_id}} + (try + (let [outcome (case type + "triples" (if attr_id + (drain! ::drain-attr-triples stop? backing-off? conn id delete-attr-triples-q {:app-id app_id :attr-id attr_id}) + (drain! ::drain-app-triples stop? backing-off? conn id delete-app-triples-q {:app-id app_id})) + "transactions" (drain! ::drain-app-transactions stop? backing-off? conn id delete-app-transactions-q {:app-id app_id}) + ;; Delete the app's attrs one at a time so each attr's triples + ;; cascade is its own bounded, committed statement. + "attrs" (drain! ::drain-app-attrs stop? backing-off? conn id delete-app-attrs-q {:app-id app_id} (attr-batch-size)) + ;; Terminal steps run to completion; they aren't interrupted mid-way. + "attr" (do (delete-attr! conn app_id attr_id) ::completed) + "app" (do (delete-app! conn app_id) ::completed))] + ;; ::stopped means the drain exited mid-way — hand the row back rather than + ;; marking a half-done step finished. + (case outcome + ::completed (finish! conn id) + ::stopped (release! conn id))) + (catch Throwable e + ;; The batch guard raises if the app/attr is no longer marked for + ;; deletion — tear the plan down. Any other failure: record it and mark + ;; the job failed so it stops being retried and can be investigated. + (if (not-deleted-error? e) + (abort-plan! conn app_id attr_id e) + (do + (tracer/record-exception-span! e {:name "custodian/process-row-error" + :attributes {:id id :type type :app-id app_id :attr-id attr_id}}) + (fail! conn id (error-string e)))))))) + +(defn tick! + "Claims and fully processes one runnable row. Returns true if it did (or + attempted) work, false when there was nothing to do." + [stop? backing-off? conn] + (if (or (flags/failing-over?) (flags/custodian-disabled?)) + false + (if-let [row (claim-row! conn)] + (do + (try + (process-row! stop? backing-off? conn row) + (catch Throwable t + ;; Leave the row leased; a stale lease lets it be retried later. + (tracer/record-exception-span! t {:name "custodian/process-row-error" + :attributes {:id (:id row) + :type (:type row) + :app-id (:app_id row)}}))) + true) + false))) + +;; ---------- +;; Worker (one per machine) + +(defonce worker (atom nil)) + +(defn- run-worker [stop? backing-off?] + (loop [] + (when-not @stop? + ;; Don't claim new work while we're backing off; drain! also pauses mid-row. + (await-capacity! stop? backing-off?) + (when-not @stop? + (let [did-work? (try + (tick! stop? backing-off? (aurora/conn-pool :write)) + (catch Throwable t + (tracer/record-exception-span! t {:name "custodian/tick-error"}) + false))] + (when-not did-work? + (Thread/sleep (long (idle-sleep-ms)))))) + (recur)))) + +(defn- reap-tick [_] + (try + (when-not (or (flags/failing-over?) (flags/custodian-disabled?)) + (reap-stuck!) + (let [reaped (reap-failed!)] + (when (seq reaped) + (tracer/record-info! {:name "custodian/reap-failed" + :attributes {:count (count reaped)}})))) + (catch Throwable t + (tracer/record-exception-span! t {:name "custodian/reap-error"})))) + +(defn start-worker [] + (let [stop? (atom false) + backing-off? (atom false) + fut (ua/vfuture (run-worker stop? backing-off?)) + reaper (chime-core/chime-at + (chime-core/periodic-seq (Instant/now) (Duration/ofHours 1)) + reap-tick) + lag-sampler (chime-core/chime-at + (chime-core/periodic-seq (Instant/now) (Duration/ofMinutes 1)) + (fn [_] (sample-lag! backing-off?)))] + (tracer/record-info! {:name "custodian/start"}) + {:stop? stop? + :backing-off? backing-off? + :future fut + :reaper reaper + :lag-sampler lag-sampler})) + +(defn stop-worker [w] + (tracer/record-info! {:name "custodian/stop"}) + (reset! (:stop? w) true) + (.close ^AutoCloseable (:reaper w)) + (.close ^AutoCloseable (:lag-sampler w)) + ;; Give the worker a chance to finish its current batch, release its claimed + ;; row, and exit before we force-cancel it. + (when (= ::timeout (deref (:future w) 10000 ::timeout)) + (future-cancel (:future w)))) + +(defn start [] + (swap! worker + (fn [w] + (or w (start-worker))))) + +(defn stop [] + (swap! worker + (fn [w] + (when w + (stop-worker w)) + nil))) + +(defn restart [] + (stop) + (start)) + +(defn before-ns-unload [] + (stop)) + +(defn after-ns-reload [] + (start)) diff --git a/server/src/instant/flags.clj b/server/src/instant/flags.clj index 6b24c38824..d3690b45e9 100644 --- a/server/src/instant/flags.clj +++ b/server/src/instant/flags.clj @@ -457,6 +457,9 @@ (defn hard-deletion-sweeper-disabled? [] (toggled? :hard-deletion-sweeper-disabled?)) +(defn custodian-disabled? [] + (toggled? :custodian-disabled?)) + (defn conn-tx-reset-disabled? [] (toggled? :conn-tx-reset-disabled?)) diff --git a/server/src/instant/hard_deletion_sweeper.clj b/server/src/instant/hard_deletion_sweeper.clj index 003a229df7..5c907c62ca 100644 --- a/server/src/instant/hard_deletion_sweeper.clj +++ b/server/src/instant/hard_deletion_sweeper.clj @@ -1,20 +1,18 @@ (ns instant.hard-deletion-sweeper (:require [chime.core :as chime-core] + [instant.custodian :as custodian] [instant.flags :as flags] - [instant.grab :as grab] - [instant.jdbc.sql :as sql] [instant.model.app :as app-model] [instant.util.date :as date-util] [instant.util.lang :as lang] [instant.util.tracer :as tracer] - [instant.db.model.attr :as attr-model] - [instant.jdbc.aurora :as aurora]) + [instant.db.model.attr :as attr-model]) (:import (java.time Duration Period ZonedDateTime))) -;; -------- -;; Config +;; ------ +;; Config (def grace-period-days 2) @@ -35,30 +33,8 @@ (def delete-timeout-seconds (.getSeconds (Duration/ofMinutes 5))) -;; ---------- -;; Sweep - -(defn straight-jacket-delete-app! [{:keys [id] :as app}] - (tracer/with-span! {:name "hard-deletion-sweeper/delete-app" - :attributes app} - (try - (binding [sql/*query-timeout-seconds* delete-timeout-seconds] - (app-model/delete-immediately-by-id! - {:id id})) - (catch Throwable e - (tracer/add-exception! e {:escaping? false}))))) - -(defn straight-jacket-delete-attr! [{app-id :app_id id :id :as attr}] - (tracer/with-span! {:name "hard-deletion-sweeper/delete-attr" - :attributes attr} - (try - (binding [sql/*query-timeout-seconds* delete-timeout-seconds] - (attr-model/hard-delete-multi! - (aurora/conn-pool :write) - app-id - #{id})) - (catch Throwable e - (tracer/add-exception! e {:escaping? false}))))) +;; ---------- +;; Sweep (defn handle-sweep [_] (when-not (flags/failing-over?) @@ -74,14 +50,10 @@ (tracer/add-data! {:attributes {:attrs-count (count attrs-to-delete) :apps-count (count apps-to-delete)}}) - (doseq [{:keys [id] :as attr} attrs-to-delete] - (grab/run-once! - (format "delete-attr-%s-%s" id (date-util/numeric-date-str maximum-marked-date)) - (fn [] (straight-jacket-delete-attr! attr)))) - (doseq [{:keys [id] :as app} apps-to-delete] - (grab/run-once! - (format "delete-app-%s-%s" id (date-util/numeric-date-str maximum-marked-date)) - (fn [] (straight-jacket-delete-app! app))))))))) + (doseq [{:keys [id app_id]} attrs-to-delete] + (custodian/enqueue-attr-deletion! {:app-id app_id :attr-id id})) + (doseq [{:keys [id]} apps-to-delete] + (custodian/enqueue-app-deletion! {:app-id id}))))))) (defn start [] (tracer/record-info! {:name "app-deletion-sweeper/schedule"}) @@ -101,4 +73,3 @@ (defn restart [] (stop) (start)) - diff --git a/server/src/instant/reactive/aggregator.clj b/server/src/instant/reactive/aggregator.clj index d63d58ff29..f600c96f94 100644 --- a/server/src/instant/reactive/aggregator.clj +++ b/server/src/instant/reactive/aggregator.clj @@ -151,7 +151,7 @@ (config/get-aurora-config) slot-name))] (with-open [connection ^PgConnection connection] - (doseq [sketches (partition-all 1000 (initial-sketch-seq connection copy-sql))] + (doseq [sketches (partition-all 100 (initial-sketch-seq connection copy-sql))] (tracer/with-span! {:name "aggregator/insert-initial-sketches" :attributes {:pid process-id :slot-name slot-name}} diff --git a/server/src/instant/reactive/invalidator.clj b/server/src/instant/reactive/invalidator.clj index e5c297220a..11cdaa8071 100644 --- a/server/src/instant/reactive/invalidator.clj +++ b/server/src/instant/reactive/invalidator.clj @@ -120,7 +120,8 @@ some-changes (or (seq idents) (seq triples) - (seq attrs)) + (seq attrs) + (seq transactions)) transactions-change (first transactions) app-id (extract-app-id transactions-change)] diff --git a/server/test/instant/custodian_test.clj b/server/test/instant/custodian_test.clj new file mode 100644 index 0000000000..56a069cb9d --- /dev/null +++ b/server/test/instant/custodian_test.clj @@ -0,0 +1,448 @@ +(ns instant.custodian-test + (:require + [clojure.test :refer [deftest testing is use-fixtures]] + [instant.config :as config] + [instant.custodian :as custodian] + [instant.db.model.attr :as attr-model] + [instant.db.transaction :as tx] + [instant.fixtures :refer [with-empty-app]] + [instant.flags :as flags] + [instant.jdbc.aurora :as aurora] + [instant.jdbc.sql :as sql])) + +;; core/start launches the custodian worker, so an ambient worker would race +;; these tests (claiming/draining the rows we enqueue). Stop it for the duration +;; and drive the logic directly, then restart it. +(use-fixtures :once + (fn [f] + (custodian/stop) + (try + (f) + (finally + (custodian/start))))) + +;; ---------- +;; Helpers + +(defn custodian-rows [conn app-id] + (sql/select ::custodian-rows + conn + ["select id, type, depends_on, worker_id + from custodian where app_id = ?::uuid order by created_at" app-id])) + +(defn row-of-type [conn app-id type] + (->> (custodian-rows conn app-id) + (filter #(= type (:type %))) + first)) + +(defn count-triples [conn app-id] + (:count (sql/select-one ::count-triples conn + ["select count(*)::int as count from triples where app_id = ?::uuid" app-id]))) + +(defn count-transactions [conn app-id] + (:count (sql/select-one ::count-transactions conn + ["select count(*)::int as count from transactions where app_id = ?::uuid" app-id]))) + +(defn count-apps [conn app-id] + (:count (sql/select-one ::count-apps conn + ["select count(*)::int as count from apps where id = ?::uuid" app-id]))) + +(defn count-triples-for-attr [conn app-id attr-id] + (:count (sql/select-one ::count-attr-triples conn + ["select count(*)::int as count from triples + where app_id = ?::uuid and attr_id = ?::uuid" app-id attr-id]))) + +(defn count-attr [conn app-id attr-id] + (:count (sql/select-one ::count-attr conn + ["select count(*)::int as count from attrs + where app_id = ?::uuid and id = ?::uuid" app-id attr-id]))) + +(defn count-attrs [conn app-id] + (:count (sql/select-one ::count-attrs conn + ["select count(*)::int as count from attrs where app_id = ?::uuid" app-id]))) + +(defn mark-app-for-deletion! [conn app-id] + (sql/do-execute! ::mark-app conn + ["update apps set deletion_marked_at = now() where id = ?::uuid" app-id])) + +(defn mark-attr-for-deletion! [conn app-id attr-id] + (sql/do-execute! ::mark-attr conn + ["update attrs set deletion_marked_at = now() + where app_id = ?::uuid and id = ?::uuid" app-id attr-id])) + +(defn seed-app! + "Adds an attr and a handful of triples across two transactions, so there's + something for custodian to delete. Returns the attr id." + [app-id] + (let [conn (aurora/conn-pool :write) + attr-id (random-uuid) + ident-id (random-uuid)] + (tx/transact! conn + (attr-model/get-by-app-id app-id) + app-id + (into [[:add-attr {:id attr-id + :forward-identity [ident-id "items" "label"] + :value-type :blob :cardinality :one + :unique? false :index? false}]] + (for [i (range 5)] + [:add-triple (random-uuid) attr-id (str "item-" i)]))) + (tx/transact! conn + (attr-model/get-by-app-id app-id) + app-id + [[:add-triple (random-uuid) attr-id "item-5"]]) + attr-id)) + +(defn add-attr-with-triples! + "Adds one attr named `label` (on the `items` etype) with a few triples. + Returns the attr id." + [app-id label] + (let [attr-id (random-uuid) + ident-id (random-uuid)] + (tx/transact! (aurora/conn-pool :write) + (attr-model/get-by-app-id app-id) + app-id + (into [[:add-attr {:id attr-id + :forward-identity [ident-id "items" label] + :value-type :blob :cardinality :one + :unique? false :index? false}]] + (for [i (range 3)] + [:add-triple (random-uuid) attr-id (str label "-" i)]))) + attr-id)) + +(defn drain-all! + "Runs the worker's tick! until there's no runnable work left." + [conn] + (let [stop? (atom false) + backing-off? (atom false)] + (loop [] + (when (custodian/tick! stop? backing-off? conn) + (recur))))) + +;; ---------- +;; Enqueue + +(deftest enqueue-app-builds-the-chain + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (let [triples (row-of-type conn app-id "triples") + transactions (row-of-type conn app-id "transactions") + attrs (row-of-type conn app-id "attrs") + app-row (row-of-type conn app-id "app")] + (testing "one row per step, all unowned" + (is (= 4 (count (custodian-rows conn app-id)))) + (is (every? (comp nil? :worker_id) [triples transactions attrs app-row]))) + (testing "chain is triples <- transactions <- attrs <- app" + (is (nil? (:depends_on triples))) + (is (= (:id triples) (:depends_on transactions))) + (is (= (:id transactions) (:depends_on attrs))) + (is (= (:id attrs) (:depends_on app-row))))) + (testing "re-enqueuing an in-flight plan is an idempotent no-op" + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (is (= 4 (count (custodian-rows conn app-id))))))))) + +(deftest enqueue-attr-builds-the-chain + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write) + attr-id (seed-app! app-id)] + (custodian/enqueue-attr-deletion! conn {:app-id app-id :attr-id attr-id}) + (let [triples (row-of-type conn app-id "triples") + attr-row (row-of-type conn app-id "attr")] + (is (= 2 (count (custodian-rows conn app-id)))) + (testing "chain is triples <- attr, both scoped to the attr" + (is (nil? (:depends_on triples))) + (is (= (:id triples) (:depends_on attr-row))))))))) + +;; ---------- +;; Claim + +(deftest claim-takes-the-runnable-row-and-owns-it + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (let [claimed (custodian/claim-row! conn)] + (testing "triples runs first (depends_on is null) and gets owned" + (is (= "triples" (:type claimed))) + (is (= @config/process-id (:worker_id claimed)))) + (testing "nothing else is runnable yet: dependents blocked, triples owned" + (is (nil? (custodian/claim-row! conn))))))))) + +;; ---------- +;; End-to-end + +(deftest drains-triples-transactions-and-the-app + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (seed-app! app-id) + (mark-app-for-deletion! conn app-id) + (is (pos? (count-triples conn app-id))) + (is (pos? (count-transactions conn app-id))) + (is (pos? (count-attrs conn app-id))) + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + ;; small batch size so the drain loops over several committed batches + (binding [flags/*flag-overrides* {:custodian-batch-size 2}] + (drain-all! conn)) + (testing "everything is gone" + (is (zero? (count-triples conn app-id))) + (is (zero? (count-transactions conn app-id))) + (is (zero? (count-attrs conn app-id))) + (is (zero? (count-apps conn app-id))) + (is (empty? (custodian-rows conn app-id)))))))) + +(deftest attrs-step-deletes-every-attr-and-leaves-the-app + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (add-attr-with-triples! app-id "a") + (add-attr-with-triples! app-id "b") + (mark-app-for-deletion! conn app-id) + (is (pos? (count-attrs conn app-id))) + ;; the attrs step on its own (what the app chain runs before the app delete) + (sql/do-execute! ::insert-attrs-step conn + ["insert into custodian (app_id, type) values (?::uuid, 'attrs')" app-id]) + (binding [flags/*flag-overrides* {:custodian-attr-batch-size 1}] + (drain-all! conn)) + (testing "every attr (and its triples) is gone, the app itself survives" + (is (zero? (count-attrs conn app-id))) + (is (zero? (count-triples conn app-id))) + (is (= 1 (count-apps conn app-id))) + (is (empty? (custodian-rows conn app-id)))))))) + +(deftest attr-deletion-deletes-only-the-target-attr + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write) + attr-a (add-attr-with-triples! app-id "a") + attr-b (add-attr-with-triples! app-id "b")] + (mark-attr-for-deletion! conn app-id attr-a) + (is (pos? (count-triples-for-attr conn app-id attr-a))) + (is (pos? (count-triples-for-attr conn app-id attr-b))) + (custodian/enqueue-attr-deletion! conn {:app-id app-id :attr-id attr-a}) + (binding [flags/*flag-overrides* {:custodian-batch-size 2}] + (drain-all! conn)) + (testing "the target attr and only its triples are gone" + (is (zero? (count-triples-for-attr conn app-id attr-a))) + (is (zero? (count-attr conn app-id attr-a)))) + (testing "the other attr, its triples, the transactions, and the app all survive" + (is (pos? (count-triples-for-attr conn app-id attr-b))) + (is (pos? (count-attr conn app-id attr-b))) + (is (pos? (count-transactions conn app-id))) + (is (pos? (count-apps conn app-id)))) + (testing "the plan is cleaned up" + (is (empty? (custodian-rows conn app-id)))))))) + +;; ---------- +;; Marked-for-deletion guard + +(deftest an-unmarked-app-is-left-alone + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (seed-app! app-id) ;; deliberately NOT marked for deletion + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (let [row (custodian/claim-row! conn)] + ;; processing catches the not-marked guard, tears down the plan, and logs + (#'custodian/process-row! (atom false) (atom false) conn row) + (testing "nothing is deleted" + (is (pos? (count-triples conn app-id))) + (is (pos? (count-transactions conn app-id))) + (is (pos? (count-apps conn app-id)))) + (testing "the deletion plan is torn down" + (is (empty? (custodian-rows conn app-id))))))))) + +(deftest an-unmarked-attr-is-left-alone + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write) + attr-id (seed-app! app-id)] ;; attr deliberately NOT marked for deletion + (custodian/enqueue-attr-deletion! conn {:app-id app-id :attr-id attr-id}) + (let [row (custodian/claim-row! conn)] + ;; processing catches the not-marked guard, tears down the plan, and logs + (#'custodian/process-row! (atom false) (atom false) conn row) + (testing "the attr and its triples survive" + (is (pos? (count-triples-for-attr conn app-id attr-id))) + (is (pos? (count-attr conn app-id attr-id)))) + (testing "the deletion plan is torn down" + (is (empty? (custodian-rows conn app-id))))))))) + +;; ---------- +;; Stop + +(deftest stopping-mid-drain-releases-the-row-and-keeps-progress + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (seed-app! app-id) + (mark-app-for-deletion! conn app-id) + (let [total (count-triples conn app-id)] + (is (> total 1)) + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (let [row (custodian/claim-row! conn) + ;; drain! reads @stop? once per batch: false on the first check + ;; (do one batch), true on the second (stop before the next). + checks (atom 0) + stop? (reify clojure.lang.IDeref + (deref [_] (> (swap! checks inc) 1)))] + (is (= "triples" (:type row))) + (binding [flags/*flag-overrides* {:custodian-batch-size 1}] + (#'custodian/process-row! stop? (atom false) conn row)) + (testing "one batch ran, then it stopped" + (is (= (dec total) (count-triples conn app-id)))) + (testing "the row is released (not finished) so it can be reclaimed" + (let [triples-row (row-of-type conn app-id "triples")] + (is (some? triples-row)) + (is (nil? (:worker_id triples-row))))) + (testing "a fresh worker reclaims it and finishes the deletion" + (drain-all! conn) + (is (zero? (count-triples conn app-id))) + (is (zero? (count-apps conn app-id)))))))))) + +;; ---------- +;; Ownership / reaper + +(deftest reaper-frees-rows-with-a-stale-owner + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + ;; Seed a row owned by a worker that stopped heartbeating. We INSERT it + ;; with an old updated_at directly: a normal UPDATE would fire the + ;; update_updated_at trigger and reset it to now (which is exactly why a + ;; live worker's row never looks stale, and a dead one's does). + (sql/do-execute! ::insert-stuck conn + ["insert into custodian (app_id, type, status, worker_id, updated_at) + values (?::uuid, 'triples', 'working', 'stuck-worker', now() - interval '10 minutes')" app-id]) + (custodian/reap-stuck! conn) + (is (nil? (:worker_id (row-of-type conn app-id "triples"))) + "a stale owner is cleared so the row can be reclaimed"))))) + +(deftest reaper-deletes-a-long-failed-job-and-its-dependents + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write) + root (random-uuid) + dep (random-uuid)] + ;; A failed root two days old, plus a dependent still waiting on it. INSERT + ;; the old updated_at directly (an UPDATE would fire the update_updated_at + ;; trigger). The dependent's own updated_at is now: it's not itself failed, + ;; but it must go too so `depends_on`'s SET NULL can't make it runnable. + (sql/do-execute! ::insert-failed conn + ["insert into custodian (id, app_id, type, status, updated_at) + values (?::uuid, ?::uuid, 'triples', 'failed', now() - interval '2 days')" root app-id]) + (sql/do-execute! ::insert-dep conn + ["insert into custodian (id, app_id, type, depends_on) + values (?::uuid, ?::uuid, 'transactions', ?::uuid)" dep app-id root]) + (is (= 2 (count (custodian-rows conn app-id)))) + (custodian/reap-failed! conn) + (testing "the whole job is gone: the failed row and its dependent" + (is (empty? (custodian-rows conn app-id)))) + (testing "a plan that failed recently is left alone" + (sql/do-execute! ::insert-recent conn + ["insert into custodian (app_id, type, status, updated_at) + values (?::uuid, 'triples', 'failed', now())" app-id]) + (custodian/reap-failed! conn) + (is (= 1 (count (custodian-rows conn app-id))))))))) + +(deftest drain-throws-if-we-no-longer-own-the-row + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (seed-app! app-id) + (mark-app-for-deletion! conn app-id) + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (let [row (custodian/claim-row! conn)] + ;; another worker takes it over + (sql/do-execute! ::steal conn + ["update custodian set worker_id = 'someone-else' where id = ?::uuid" (:id row)]) + ;; the heartbeat's ownership guard raises, unwinding the drain. (process-row! + ;; deliberately swallows this to fail the job, so we test drain! directly.) + (is (thrown? Exception + (#'custodian/drain! ::drain (atom false) (atom false) conn (:id row) + custodian/delete-app-triples-q {:app-id app-id})))))))) + +;; ---------- +;; Backpressure + +(deftest sample-lag-sets-backing-off-against-the-threshold + (testing "over the 100mb default -> back off" + (let [backing-off? (atom false)] + (with-redefs [custodian/max-replication-lag-bytes (fn [_] (* 200 1024 1024))] + (custodian/sample-lag! backing-off?)) + (is (true? @backing-off?)))) + (testing "under the threshold -> don't back off" + (let [backing-off? (atom true)] + (with-redefs [custodian/max-replication-lag-bytes (fn [_] 0)] + (custodian/sample-lag! backing-off?)) + (is (false? @backing-off?))))) + +;; ---------- +;; Failover / disabled pause + +(deftest should-pause-covers-lag-failover-and-disabled + (let [not-lagging (atom false)] + (testing "no lag and no flags set -> don't pause" + (is (not (#'custodian/should-pause? not-lagging)))) + (testing "replication lag -> pause" + (is (#'custodian/should-pause? (atom true)))) + (testing "failing over -> pause even without lag" + (binding [flags/*toggle-overrides* {:failing-over true}] + (is (#'custodian/should-pause? not-lagging)))) + (testing "custodian disabled -> pause even without lag" + (binding [flags/*toggle-overrides* {:custodian-disabled? true}] + (is (#'custodian/should-pause? not-lagging)))))) + +(deftest failing-over-parks-an-in-flight-drain-then-resumes + (with-empty-app + (fn [app] + (let [app-id (:id app) + conn (aurora/conn-pool :write)] + (seed-app! app-id) + (mark-app-for-deletion! conn app-id) + (let [total (count-triples conn app-id) + ;; Delivered the moment the drain hits its park loop, so the + ;; assertions below run deterministically without a fixed sleep. + parked (promise)] + (is (pos? total)) + (custodian/enqueue-app-deletion! conn {:app-id app-id}) + (let [row (custodian/claim-row! conn) + stop? (atom false)] + (is (= "triples" (:type row))) + ;; The idle sleep both signals that the drain reached the park and + ;; keeps the interval tiny so stopping unparks it right away. + (with-redefs [custodian/idle-sleep-ms (fn [] (deliver parked true) 1)] + (let [fut (future + (binding [flags/*toggle-overrides* {:failing-over true}] + (#'custodian/process-row! stop? (atom false) conn row)))] + (try + (testing "while failing over the drain parks and deletes nothing" + (is (true? (deref parked 2000 ::timeout)) "drain reached its park loop") + (is (not (realized? fut))) + (is (= total (count-triples conn app-id)))) + (testing "stopping unparks it; the untouched row is handed back" + (reset! stop? true) + (is (not= ::timeout (deref fut 5000 ::timeout)) "drain returned after stop") + (let [triples-row (row-of-type conn app-id "triples")] + (is (some? triples-row)) + (is (nil? (:worker_id triples-row))))) + (finally + (reset! stop? true) + (future-cancel fut))))) + (testing "once we're no longer failing over the deletion finishes" + (drain-all! conn) + (is (zero? (count-triples conn app-id))) + (is (zero? (count-apps conn app-id)))))))))) diff --git a/server/test/instant/db/app_backup_jobs_test.clj b/server/test/instant/db/app_backup_jobs_test.clj index 64ceb9d6ed..fd8b322158 100644 --- a/server/test/instant/db/app_backup_jobs_test.clj +++ b/server/test/instant/db/app_backup_jobs_test.clj @@ -92,9 +92,11 @@ (testing "we never exceeded n concurrent backups" (is (= n @max-running)))) (finally - ;; Make sure nothing stays blocked if an assertion threw. + ;; Make sure nothing stays blocked if an assertion threw, and wait + ;; for the pool to fully terminate so no straggler worker drains the + ;; shared table in the next test. (deliver release true) - (.shutdown pool)))))))) + (app-backup-jobs/shutdown-pool pool wait-timeout)))))))) (deftest rejects-a-second-in-flight-backup-for-the-same-app (let [pool (app-backup-jobs/make-pool 1) @@ -116,8 +118,10 @@ (is (re-find #"already in progress" (:instant.util.exception/message err)))) (finally + ;; Wait for the pool to fully terminate so this test's worker can't + ;; still be draining the shared table when the next test starts. (deliver release true) - (.shutdown pool)))))))) + (app-backup-jobs/shutdown-pool pool wait-timeout)))))))) (deftest killing-the-pool-cancels-in-flight-work ;; Just the pool mechanics--no db, no real backup. We run a task shaped like @@ -170,4 +174,4 @@ "the backup vfuture unwound its resources") (finally (deliver block true) - (.shutdownNow pool))))) + (app-backup-jobs/shutdown-pool pool wait-timeout)))))