Conversation
📝 WalkthroughWalkthroughThe change adds a database-backed custodian worker for dependency-ordered app and attribute deletion. It supports leases, bounded cleanup, retries, replication-lag backpressure, lifecycle control, sweeper integration, workflow tests, and reactive processing updates. ChangesCustodian deletion processing
Reactive change processing
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Application
participant HardDeletionSweeper
participant Custodian
participant Database
Application->>Custodian: start worker
HardDeletionSweeper->>Custodian: enqueue app or attribute deletion
Custodian->>Database: claim dependency-ready row
Custodian->>Database: delete bounded data batch
Database-->>Custodian: commit and clear dependency
Application->>Custodian: stop worker
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (6)
server/src/instant/custodian.clj (4)
217-219: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid
returning :*when only the count is needed.The
deletedCTE returns every column of every deleted row. For a batch ofbatch-sizetriples this materializes wide rows for no reason. Return a constant instead.♻️ Proposed change
[:deleted {:delete-from table :where [:in :ctid {:select :ctid :from :to-delete}] - :returning :*}]] + :returning [[:inline 1] :one]}]]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 217 - 219, Update the deleted CTE definition to return a constant value instead of all columns, replacing :returning :* while preserving the existing delete condition and CTE structure.
11-16: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUpdate the docstring: rows do not have
done_at.The schema in
server/resources/migrations/124_add_custodian.up.sqlhas nodone_atcolumn. A step finishes by deleting its row, and theon delete set nullFK clears the dependent'sdepends_on. The docstring at Line 12 and Line 14 still describes adone_atstate.📝 Proposed doc fix
- `depends_on` points at the step that must finish first, so a row is runnable - once the row it depends on is `done_at`. Each worker claims one runnable row - with a lease, drains it in `batch-size` chunks (each chunk its own committed - transaction), then marks it done — which makes its dependent runnable. The + `depends_on` points at the step that must finish first, so a row is runnable + once `depends_on` is null. Each worker claims one runnable row with a lease, + drains it in `batch-size` chunks (each chunk its own committed transaction), + then deletes its row; the `on delete set null` FK clears `depends_on` on the + dependent, which makes the dependent runnable. The🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 11 - 16, Update the custodian docstring to remove all references to rows becoming runnable through a done_at state. Describe that completing a step deletes its row, the on-delete-set-null foreign key clears the dependent row’s depends_on, and the dependent then becomes runnable; retain the existing lease, batching, transaction, and cascade behavior.
332-342: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winDo not put side effects inside
swap!.
swap!can retry its function when the atom is contended. A retry would callstart-workertwice and leak the first worker's future and chime schedulers.stophas the mirror problem: it could callstop-workertwice. Uselockingon the atom instead.♻️ Proposed change
(defn start [] - (swap! worker - (fn [w] - (or w (start-worker))))) + (locking worker + (when-not `@worker` + (reset! worker (start-worker))))) (defn stop [] - (swap! worker - (fn [w] - (when w - (stop-worker w)) - nil))) + (locking worker + (when-let [w `@worker`] + (stop-worker w)) + (reset! worker nil)))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 332 - 342, Replace the side-effecting swap! callbacks in start and stop with locking on the worker atom, performing start-worker or stop-worker exactly once while holding the lock and updating the atom explicitly. Preserve start’s existing-worker reuse and stop’s nil reset behavior.
253-259: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueAdd a default branch to the
caseontype.
casewithout a default throwsIllegalArgumentExceptionfor an unrecognizedtype.tick!then swallows it and the row stays leased until the reaper frees it, which repeats forever. An explicit default makes the failure self-describing.♻️ Proposed change
"attr" (attr-model/hard-delete-multi! conn app_id #{attr_id}) - "app" (app-model/delete-immediately-by-id! conn {:id app_id})) + "app" (app-model/delete-immediately-by-id! conn {:id app_id}) + (throw (ex-info "custodian: unknown row type" {:id id :type type :app-id app_id})))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 253 - 259, Add an explicit default branch to the case expression handling type in tick!, throwing a descriptive error that includes the unrecognized type. Preserve the existing branches for "triples", "transactions", "attr", and "app".server/test/instant/custodian_test.clj (2)
141-143: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe assertions do not check the claim in the
testinglabel.The label states "both scoped to the attr", but no assertion inspects
attr_id. Thecustodian-rowshelper at Lines 27-31 does not even selectattr_id. Addattr_idto the helper and assert it on both rows.💚 Proposed change
- ["select id, type, depends_on, worker_id + ["select id, type, attr_id, depends_on, worker_id from custodian where app_id = ?::uuid order by created_at" app-id]))(testing "chain is triples <- attr, both scoped to the attr" (is (nil? (:depends_on triples))) - (is (= (:id triples) (:depends_on attr-row))))))))) + (is (= (:id triples) (:depends_on attr-row))) + (is (= attr-id (:attr_id triples))) + (is (= attr-id (:attr_id attr-row)))))))))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/test/instant/custodian_test.clj` around lines 141 - 143, Update the custodian-rows helper to select attr_id, then extend the “chain is triples <- attr, both scoped to the attr” test to assert attr_id on both triples and attr-row. Preserve the existing depends_on and id assertions.
16-22: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRestart the worker only if it was running.
The fixture always calls
custodian/startin thefinally. If the worker was not running when this namespace started, the fixture leaves a live worker behind. That worker then claims and drains custodian rows created by other test namespaces, which reintroduces the exact race the fixture avoids. Capture the prior state and restore it.♻️ Proposed change
(use-fixtures :once (fn [f] - (custodian/stop) - (try - (f) - (finally - (custodian/start))))) + (let [was-running? (some? `@custodian/worker`)] + (custodian/stop) + (try + (f) + (finally + (when was-running? + (custodian/start)))))))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/test/instant/custodian_test.clj` around lines 16 - 22, Update the use-fixtures setup to capture whether the custodian was running before calling custodian/stop, then in the finally block call custodian/start only when that prior state was running; otherwise leave it stopped.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@server/src/instant/core.clj`:
- Around line 363-365: Update the shutdown hook in core/-main to call
hard-deletion-sweeper/stop before starting the parallel shutdown tasks, ensuring
the producer is closed before custodian/stop runs. Use the existing
hard-deletion-sweeper lifecycle cleanup function and preserve the current
shutdown tracing behavior.
In `@server/src/instant/custodian.clj`:
- Line 238: Remove the leftover `(tool/def-locals)` debug call from the
custodian namespace initialization, and remove the now-unused `tool` alias from
the `ns` declaration. Ensure instant.custodian loads without referencing the
nonexistent namespace.
- Around line 312-314: Update the reaper schedule in the custodian startup flow
around reap-tick and chime-core/periodic-seq so its interval is derived from the
configured lease timeout instead of a fixed one-hour Duration. Reuse the
existing lease-timeout configuration symbol and preserve the current periodic
reaping behavior.
- Around line 178-184: Update sample-lag! to call max-replication-lag-bytes with
the write pool from aurora/conn-pool :write instead of the read pool, while
preserving the existing threshold, reset!, and exception-recording behavior.
In `@server/src/instant/hard_deletion_sweeper.clj`:
- Around line 77-80: Update handle-sweep around the attrs-to-delete and
apps-to-delete doseq loops to catch enqueue exceptions per entity, record each
failure, and continue processing subsequent attributes and applications.
Preserve custodian/enqueue-attr-deletion! and custodian/enqueue-app-deletion!
behavior, including duplicate suppression and propagation of non-duplicate
database errors within the enqueue functions.
---
Nitpick comments:
In `@server/src/instant/custodian.clj`:
- Around line 217-219: Update the deleted CTE definition to return a constant
value instead of all columns, replacing :returning :* while preserving the
existing delete condition and CTE structure.
- Around line 11-16: Update the custodian docstring to remove all references to
rows becoming runnable through a done_at state. Describe that completing a step
deletes its row, the on-delete-set-null foreign key clears the dependent row’s
depends_on, and the dependent then becomes runnable; retain the existing lease,
batching, transaction, and cascade behavior.
- Around line 332-342: Replace the side-effecting swap! callbacks in start and
stop with locking on the worker atom, performing start-worker or stop-worker
exactly once while holding the lock and updating the atom explicitly. Preserve
start’s existing-worker reuse and stop’s nil reset behavior.
- Around line 253-259: Add an explicit default branch to the case expression
handling type in tick!, throwing a descriptive error that includes the
unrecognized type. Preserve the existing branches for "triples", "transactions",
"attr", and "app".
In `@server/test/instant/custodian_test.clj`:
- Around line 141-143: Update the custodian-rows helper to select attr_id, then
extend the “chain is triples <- attr, both scoped to the attr” test to assert
attr_id on both triples and attr-row. Preserve the existing depends_on and id
assertions.
- Around line 16-22: Update the use-fixtures setup to capture whether the
custodian was running before calling custodian/stop, then in the finally block
call custodian/start only when that prior state was running; otherwise leave it
stopped.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: f10a412d-d77e-4a66-b58d-0898d8ac8daf
📒 Files selected for processing (7)
server/resources/migrations/124_add_custodian.down.sqlserver/resources/migrations/124_add_custodian.up.sqlserver/src/instant/core.cljserver/src/instant/custodian.cljserver/src/instant/flags.cljserver/src/instant/hard_deletion_sweeper.cljserver/test/instant/custodian_test.clj
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
server/src/instant/custodian.clj (2)
353-363: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winKeep worker lifecycle side effects outside
swap!update functions.Clojure may retry
swap!update functions, sostart-workercan create an extra worker/scheduler that is not stored in the atom, andstop-workercan close the same worker resources more than once. Guardstartandstopso only one lookup/update happens after no worker is present/active, and create or close resources only once.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 353 - 363, The start and stop lifecycle side effects must not run inside the retryable swap! functions. Update start and stop to perform an atomic worker-state check/update first, then call start-worker only when transitioning from no worker and stop-worker only once when transitioning from an active worker, ensuring created or closed resources are not duplicated during swap! retries.
40-45: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winAdd lower-bound validation for custodian runtime flags.
:custodian-batch-sizeis passed directly as the delete query:limit; zero or negative values select no rows, makedrain!return::completed, and causeprocess-row!to finish without deleting work. Reject non-positive batch and lease values, and reject negative idle sleep.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 40 - 45, Add lower-bound validation to the flag accessors batch-size, idle-sleep-ms, and lease-timeout-seconds: require batch size and lease timeout to be positive, and idle sleep to be non-negative. Ensure invalid runtime flag values are rejected before they reach drain!, process-row!, or the delete query.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@server/src/instant/custodian.clj`:
- Around line 353-363: The start and stop lifecycle side effects must not run
inside the retryable swap! functions. Update start and stop to perform an atomic
worker-state check/update first, then call start-worker only when transitioning
from no worker and stop-worker only once when transitioning from an active
worker, ensuring created or closed resources are not duplicated during swap!
retries.
- Around line 40-45: Add lower-bound validation to the flag accessors
batch-size, idle-sleep-ms, and lease-timeout-seconds: require batch size and
lease timeout to be positive, and idle sleep to be non-negative. Ensure invalid
runtime flag values are rejected before they reach drain!, process-row!, or the
delete query.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: eed0800f-cd1a-4431-9929-5af2410e6b74
📒 Files selected for processing (1)
server/src/instant/custodian.clj
|
View Vercel preview at instant-www-js-custodian-jsv.vercel.app. |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
server/test/instant/custodian_test.clj (2)
322-332: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winTest the replication-lag threshold boundary.
The values
200 MiBand0do not verify the configured100 MBboundary. A much lower threshold would still pass both cases. Add assertions for the configured threshold and one byte above it.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/test/instant/custodian_test.clj` around lines 322 - 332, Update sample-lag-sets-backing-off-against-the-threshold to test custodian’s configured 100 MB boundary directly: assert no backoff at exactly the threshold and backoff at one byte above it, using max-replication-lag-bytes values that distinguish these adjacent cases.
304-317: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winAssert the ownership-loss failure specifically.
Line 316 accepts any
Exception. A database failure or an unrelated processing error would also pass this test. Assert the expected ownership-specific exception type and error data or message.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/test/instant/custodian_test.clj` around lines 304 - 317, Update the assertion around the private custodian/process-row! call in drain-throws-if-we-no-longer-own-the-row to require the specific ownership-loss exception type and verify its expected error data or message, rather than accepting any Exception. Keep the existing setup that changes worker_id to someone-else and preserve the test’s focus on ownership loss.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@server/test/instant/custodian_test.clj`:
- Around line 322-332: Update sample-lag-sets-backing-off-against-the-threshold
to test custodian’s configured 100 MB boundary directly: assert no backoff at
exactly the threshold and backoff at one byte above it, using
max-replication-lag-bytes values that distinguish these adjacent cases.
- Around line 304-317: Update the assertion around the private
custodian/process-row! call in drain-throws-if-we-no-longer-own-the-row to
require the specific ownership-loss exception type and verify its expected error
data or message, rather than accepting any Exception. Keep the existing setup
that changes worker_id to someone-else and preserve the test’s focus on
ownership loss.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 3311cb02-2dff-4ba1-8249-dac751dfaa52
📒 Files selected for processing (2)
server/src/instant/custodian.cljserver/test/instant/custodian_test.clj
🚧 Files skipped from review as they are similar to previous changes (1)
- server/src/instant/custodian.clj
dwwoelfel
changed the title
dwwoelfel
changed the title
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
server/src/instant/reactive/invalidator.clj (1)
123-130: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRead transaction DELETE data from
:identity.When the custodian deletes rows from
transactions, the old row is available in:identity, not:columns. The current extraction functions read only:columns, soapp-idcan be nil and thewhenguard drops the WAL record before invalidation. The transaction ID and creation timestamp have the same problem.Select
:identityfor:deletechanges and use that source consistently for all transaction metadata.Proposed fix
+(defn transaction-columns [{:keys [action columns identity]}] + (if (= :delete action) + identity + columns)) + (defn extract-app-id - [{:keys [columns] :as _change}] - (app-id-from-columns columns)) + [change] + (app-id-from-columns (transaction-columns change))) (defn extract-tx-id [{:keys [columns] :as _change}] - (topics/get-column columns "id")) + (topics/get-column (transaction-columns _change) "id")) (defn extract-tx-created-at [{:keys [columns] :as _change}] - (when-let [^String created-at (topics/get-column columns "created_at")] + (when-let [^String created-at (topics/get-column (transaction-columns _change) "created_at")] (.toInstant (Timestamp/valueOf created-at))))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/reactive/invalidator.clj` around lines 123 - 130, Update the transaction metadata extraction around transactions-change, app-id, tx-id, and tx-created-at to select :identity for :delete changes and :columns otherwise. Use this same selected source consistently when calling extract-app-id, extract-tx-id, and extract-tx-created-at so DELETE WAL records retain their metadata and reach invalidation.server/src/instant/custodian.clj (1)
103-115: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUpdate the docstring to include the
attrsstep.
enqueue-app-deletion-qnow inserts four rows:triples <- transactions <- attrs <- app. The docstring still lists three steps.📝 Proposed fix
(defn enqueue-app-deletion! - "Idempotently enqueues the chain triples <- transactions <- app for an app, + "Idempotently enqueues the chain triples <- transactions <- attrs <- app for an app, atomically so a worker never sees a partial chain."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/src/instant/custodian.clj` around lines 103 - 115, Update the docstring for enqueue-app-deletion! to list the complete deletion chain as triples <- transactions <- attrs <- app, matching the four rows inserted by enqueue-app-deletion-q.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@server/src/instant/custodian.clj`:
- Around line 428-430: Add a deletion-marked guard to the terminal "app" and
"attr" branches in the custodian step dispatch before invoking
app-model/delete-immediately-by-id! or attr-model/hard-delete-multi!. Reuse the
existing batch restore-check mechanism and not-deleted-message behavior so a
restored target causes the terminal delete to fail rather than deleting by id.
---
Outside diff comments:
In `@server/src/instant/custodian.clj`:
- Around line 103-115: Update the docstring for enqueue-app-deletion! to list
the complete deletion chain as triples <- transactions <- attrs <- app, matching
the four rows inserted by enqueue-app-deletion-q.
In `@server/src/instant/reactive/invalidator.clj`:
- Around line 123-130: Update the transaction metadata extraction around
transactions-change, app-id, tx-id, and tx-created-at to select :identity for
:delete changes and :columns otherwise. Use this same selected source
consistently when calling extract-app-id, extract-tx-id, and
extract-tx-created-at so DELETE WAL records retain their metadata and reach
invalidation.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 263a6d00-14cd-42ba-8aba-79dc70d3b004
📒 Files selected for processing (6)
server/resources/migrations/124_add_custodian.down.sqlserver/resources/migrations/124_add_custodian.up.sqlserver/src/instant/custodian.cljserver/src/instant/reactive/aggregator.cljserver/src/instant/reactive/invalidator.cljserver/test/instant/custodian_test.clj
🚧 Files skipped from review as they are similar to previous changes (1)
- server/test/instant/custodian_test.clj
| 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))] |
There was a problem hiding this comment.
This is just to reduce memory usage when creating the aggregator from scratch.
| (seq triples) | ||
| (seq attrs)) | ||
| (seq attrs) | ||
| (seq transactions)) |
There was a problem hiding this comment.
This allows the invalidator slot to make progress when we delete a bunch of transactions that have no triples.
1 participant
Creates a new
custodianprocess that allows us to delete large apps. Previously, we had to set a very long timeout to delete large apps. That put pressure on the invalidator and would fail for very large apps.This PR will delete the triples and transactions in batches first, then finally delete the app.
Each machine runs one deletion process that polls for new work in the background. We only have one because my main concern is preventing too much load on the database. If the database can handle it and the replication lag isn't affected, we could increase the number of processes.
Each process is essentially a linked list in the database. It looks like this:
Delete app:
Delete app->Delete attrs->Delete triples->Delete transactionsDelete attr
Delete attr->Delete triplesThe worker grabs the tail of the list, performs the work, then deletes the item so the next worker can pick up the new tail of the list.
Delete transactionsandDelete triplesboth run a long-running process that deletes the rows 1000 at a time (configurable with a flag). Each time it deletes, it 1. checks to make sure the app/attr hasdeletion_marked_atand 2. updates a heartbeat on the job. There's a separate cron process that frees stuck jobs.In the background, we poll the replication slots to make sure that the lag doesn't get out of hand. If it does, we automatically pause the workers until replication catches up. We can raise or lower the lag threshold with a flag.
If a job fails, we'll try it two more times, then mark it failed. Then we'll try it again after 24 hours.