Custodian: a sweeper to delete apps in smaller chunks

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();

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();

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