feat(cluster): implement Postgres cluster plugin#4220
Open
asmith987 wants to merge 1 commit into
Open
Conversation
|
Warning Review limit reached
Next review available in: 57 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (47)
📝 WalkthroughWalkthroughChangesThe PR adds a PostgreSQL cluster plugin with SQL-backed cache, advisory locks, TTL reapers, LISTEN/NOTIFY watches, configuration, providers, lifecycle handling, migrations, integration tests, and documentation. It also refactors cluster conformance runners to support per-scenario async factories, teardown, and virtual or real time. Postgres cluster plugin
Discovery polling and conformance runtime
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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: 13
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
gears/system/cluster/cluster-conformance/src/cache.rs (1)
321-329: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBound asynchronous watch waits in the cache conformance suite.
wait_forawaits each receive indefinitely, so notification or polling regressions hang the suite instead of producing assertion failures.
gears/system/cluster/cluster-conformance/src/cache.rs#L321-L329: wrap the expiry-event wait in a bounded timeout.gears/system/cluster/cluster-conformance/src/cache.rs#L433-L473: bound both theChangedandDeletedwaits.🤖 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 `@gears/system/cluster/cluster-conformance/src/cache.rs` around lines 321 - 329, Bound the asynchronous waits in cache.rs: at lines 321-329, wrap the expiry-event wait around wait_for and at lines 433-473, wrap both Changed and Deleted waits with the existing timeout mechanism. Ensure timeout failures become assertion failures rather than allowing the conformance suite to hang, while preserving the current event predicates and success behavior.
🟡 Minor comments (9)
gears/system/cluster/cluster-conformance/src/lock.rs-145-150 (1)
145-150: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winLeave more margin before renewing the real-time lease.
Line 146 leaves only 50 ms for scheduling and the database round trip. Under CI load, the 200 ms lease can expire before
renew, making this conformance test flaky. Elapsing 50 ms still verifies pre-expiry renewal with a safer margin.Proposed fix
- time.elapse(Duration::from_millis(150)).await; + time.elapse(Duration::from_millis(50)).await;🤖 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 `@gears/system/cluster/cluster-conformance/src/lock.rs` around lines 145 - 150, Adjust the timing in the SC-LOCK-005 renewal test so the lease is renewed with a larger safety margin before expiry: reduce the elapsed duration in the time.elapse call preceding guard.renew from 150 ms to 50 ms, while preserving the 200 ms lease and existing success assertion.gears/system/cluster/cluster-conformance/src/time.rs-65-72 (1)
65-72: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDocument the capped real-time semantics accurately.
Line 65 promises that at least
durpasses, butRealsleeps for at most 500 ms. Clarify this contract to avoid writing invalid timing assertions.Proposed fix
- /// Makes (at least) `dur` of scenario time pass, then yields so background - /// reaper / renewal / sweeper tasks get a chance to run: instant - /// [`tokio::time::advance`] under [`Virtual`](Self::Virtual), a real (capped) - /// [`tokio::time::sleep`] under [`Real`](Self::Real). + /// Advances scenario time by `dur` under [`Virtual`](Self::Virtual), or + /// sleeps for `min(dur, REAL_CAP)` under [`Real`](Self::Real), then yields + /// so background reaper / renewal / sweeper tasks can run.🤖 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 `@gears/system/cluster/cluster-conformance/src/time.rs` around lines 65 - 72, Update the documentation for Time::elapse to state that Virtual advances by dur, while Real sleeps for the capped duration dur.min(Self::REAL_CAP), so the Real variant does not promise that the full requested duration elapses.gears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rs-101-125 (1)
101-125: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winSynchronize on waiter registration before releasing.
!waiter.is_finished()does not prove the task reachedReleaseWaiters; an unscheduled task can start after release, acquire immediately, and satisfy the latency assertion without exercising NOTIFY. Add a test-only registration barrier or observable waiter count before releasing.🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rs` around lines 101 - 125, The integration test must synchronize with the waiter’s registration before calling guard.release(), because waiter.is_finished() only confirms the task is running, not that lock() reached ReleaseWaiters. Update the waiter setup around lock() to expose a test-only registration barrier or observable waiter count, await that confirmation, then release the guard so the latency assertion specifically exercises the NOTIFY wake path.gears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rs-46-48 (1)
46-48: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winUse the standard watch-delivery timeout here.
A 500 ms deadline is substantially tighter than the neighboring 2–3 second LISTEN/NOTIFY tests and can fail under loaded CI despite correct delivery.
Proposed fix
- let event = tokio::time::timeout(Duration::from_millis(500), watch.recv()) + let event = tokio::time::timeout(Duration::from_secs(3), watch.recv())🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rs` around lines 46 - 48, Update the timeout wrapping watch.recv() in the event delivery test to use the standard watch-delivery timeout established by the neighboring LISTEN/NOTIFY tests, replacing the 500 ms deadline while preserving the existing timeout failure assertion.gears/system/cluster/plugins/postgres-cluster-plugin/tests/common/mod.rs-290-303 (1)
290-303: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winMake
timeoutbound each pending condition too.The deadline is checked only after
condition().awaitreturns, so stalled SQL or network I/O can exceed the requested timeout indefinitely.Proposed fix
pub async fn wait_until<F>(timeout: Duration, interval: Duration, mut condition: F) -> bool where F: AsyncFnMut() -> bool, { - let deadline = tokio::time::Instant::now() + timeout; - loop { - if condition().await { - return true; - } - if tokio::time::Instant::now() >= deadline { - return false; + tokio::time::timeout(timeout, async { + loop { + if condition().await { + return true; + } + tokio::time::sleep(interval).await; } - tokio::time::sleep(interval).await; - } + }) + .await + .unwrap_or(false) }🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/tests/common/mod.rs` around lines 290 - 303, Update wait_until so each condition().await is bounded by the overall deadline, preventing stalled SQL or network I/O from exceeding timeout indefinitely. Compute the remaining duration before invoking condition and apply it to the await (for example, with Tokio’s timeout); return false when the condition exceeds the deadline, while preserving the existing success and polling behavior.gears/system/cluster/plugins/postgres-cluster-plugin/src/config.rs-101-112 (1)
101-112: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winReject zero cache and lock reaper intervals before startup.
cache_reaper_interval_msandlock_reaper_interval_msflow straight intotokio::time::interval, which panics on a zero period. Add a non-zero validation or make the config reject0here.🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/src/config.rs` around lines 101 - 112, Validate cache_reaper_interval_ms and lock_reaper_interval_ms as non-zero during configuration deserialization or startup before they reach tokio::time::interval. Update the configuration validation associated with these fields, preserving their existing defaults and rejecting zero values with a clear configuration error.gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rs-179-195 (1)
179-195: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winTreat expired rows as absent in delete operations.
Both deletes can remove an expired-but-unreaped row, return
true, and emitDeleted, despiteget,contains, and CAS treating that row as absent. Add the live-entry predicate to both statements.Proposed correction
-DELETE FROM {table} WHERE key = $1 RETURNING 1 +DELETE FROM {table} +WHERE key = $1 + AND (expires_at IS NULL OR expires_at > now()) +RETURNING 1 -DELETE FROM {table} WHERE key = $1 AND value = $2 RETURNING 1 +DELETE FROM {table} +WHERE key = $1 + AND value = $2 + AND (expires_at IS NULL OR expires_at > now()) +RETURNING 1Also applies to: 298-325
🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rs` around lines 179 - 195, Update both delete SQL statements in the `delete` implementations to include the same live-entry expiration predicate used by `get`, `contains`, and CAS, so expired rows are neither removed nor reported as deleted. Preserve the existing transaction, notification, and boolean-result behavior for live rows.gears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.md-59-60 (1)
59-60: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUse the implemented 7997-byte key limit consistently.
These sections still specify 8190 bytes, while
MAX_KEY_BYTESandDESIGN.mdenforce 7997. Update the test descriptions so boundary coverage cannot drift again.Also applies to: 140-140, 259-260
🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.md` around lines 59 - 60, Update the cache/watch.rs test descriptions to use the implemented 7997-byte key limit instead of 8190, including the corresponding occurrences at the additional referenced sections. Keep the descriptions aligned with MAX_KEY_BYTES and DESIGN.md so the documented boundary coverage remains consistent.gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs-113-145 (1)
113-145: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winRemove the overflow → empty-payload claim. PostgreSQL queue exhaustion aborts the committing transaction; it does not emit an empty notification payload. Keep
ParsedNotification::Resetfor empty/malformed/external payloads, but rewrite the docs/tests so overflow is described as a commit failure instead.
gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs#L113-L145: keep the fallback, but drop the overflow wording from the doc comment.gears/system/cluster/plugins/postgres-cluster-plugin/docs/DESIGN.md#L123-L123and#L388-L390: replace the overflow-basedResetcontract with commit-failure / queue-usage language.gears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.md#L259-L260and#L279-L279: retitle these cases around explicit empty-payload handling or commit-failure behavior.🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs` around lines 113 - 145, Remove the claim that PostgreSQL queue overflow produces an empty payload while preserving ParsedNotification::Reset for empty, malformed, and external payloads. In gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs lines 113-145, update the parse_notification documentation; in docs/DESIGN.md lines 123 and 388-390, describe overflow as a commit failure and clarify queue usage; in docs/TESTING.md lines 259-260 and 279, retitle the cases for explicit empty-payload handling or commit-failure behavior.
🧹 Nitpick comments (1)
gears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rs (1)
20-23: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAlign the overview with the implemented lock isolation.
This says lock scenarios use fresh containers, but
lock_conformancenow shares one container and relies on per-schema tables plusdrain_held.Proposed fix
-//! suite goes further and hands out a fresh **container** per scenario, since -//! `pg_advisory_lock`'s key space is server-wide, not schema-scoped (see -//! `lock_conformance`). +//! suite also uses per-scenario schemas on one shared container; teardown +//! drains held advisory locks before constructing the next scenario (see +//! `lock_conformance`).🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rs` around lines 20 - 23, Update the module overview near lock_conformance to describe the implemented isolation: scenarios share one container and use per-schema tables plus drain_held, rather than receiving a fresh container per scenario. Keep the existing search_path and server-wide advisory-lock context accurate.
🤖 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 `@gears/system/cluster/cluster-conformance/src/factory.rs`:
- Around line 57-70: Update run_scenario so the run callback executes behind an
async panic boundary, ensuring scenario_backend.teardown is awaited even when
run panics. After teardown completes, resume the captured panic so the original
assertion failure remains visible; preserve the existing build, run, and
teardown ordering for successful scenarios.
In `@gears/system/cluster/cluster/src/defaults/discovery.rs`:
- Around line 632-662: Update reconcile_view and the shared registry state to
track a per-service generation/revision. Capture the service revision before the
asynchronous scan/get sweep, then under the lock apply the rebuilt map only if
that revision is unchanged; otherwise discard the stale sweep or retry it.
Ensure StoreMaintainer::apply increments the same service revision for joins,
deletions, and state changes so newer watch events cannot be overwritten.
In `@gears/system/cluster/cluster/src/wiring.rs`:
- Around line 307-317: Update the service-discovery auto-wrap path in the
cluster wiring to resolve the configured per-profile sd_poll_interval_ms and
pass it to with_prefix_watch_polling. Preserve the existing default only when no
profile-specific value is configured, ensuring operator-provided polling cadence
is honored instead of the fixed DEFAULT_PREFIX_WATCH_POLL.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/Cargo.toml`:
- Around line 1-13: Add the required cf-studio-path package metadata to the
Cargo.toml manifest, setting it to ".cf-studio" alongside the existing package
configuration in [package].
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rs`:
- Around line 68-81: Update ttl_to_expires_at so finite TTLs are represented as
a validated duration for SQL rather than calculated with chrono::Utc::now().
Bind that duration in the insert/upsert query and compute expires_at using
PostgreSQL now() plus the bound interval, while preserving None for
Ttl::Indefinite and the existing out-of-range validation.
- Around line 33-40: The Postgres cache currently embeds the configured schema
directly into query identifiers. In PostgresCache::new, validate or safely quote
the schema once when constructing the table identifier, then update reaper.rs at
lines 32-36 to consume only that validated identifier rather than rebuilding or
using the raw schema; apply the corresponding change in mod.rs lines 33-40 and
reaper.rs lines 32-36.
In
`@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch_tests.rs`:
- Around line 83-120: Update the reset and shutdown tests around
dispatch_reset_broadcasts_to_every_key_and_clears_subscriptions and
close_all_sends_terminal_shutdown_to_every_watcher to fill each watcher's
64-slot buffer before invoking registry.dispatch(Reset) or registry.close_all().
Assert that Reset and Closed(ClusterError::Shutdown) are still received as the
required terminal events, rather than only observing channel closure.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs`:
- Around line 252-270: The broadcast_and_clear path can drop terminal Reset or
Closed events when a watcher channel is full. Update broadcast_and_clear to
guarantee delivery of the event to every watcher by reserving channel capacity
or asynchronously awaiting delivery before removing senders, and only clear
subscriptions after each event has been delivered. Preserve the existing
per-watcher fresh-event behavior and the terminal Closed(ClusterError::Shutdown)
behavior in close_all.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/mod.rs`:
- Around line 210-217: Update the lifecycle handling around try_acquire, stop,
and drain_held to serialize shutdown against acquisitions: add a
shared/exclusive lifecycle gate, acquire shared access for the full acquisition
path, and acquire exclusive access before cancelling and draining. Reject new
acquisitions once shutdown begins, ensuring no lock can be inserted after the
drain snapshot and pool closure cannot be blocked by a stranded guard.
- Around line 101-105: Use acquisition holder_id as the end-to-end ownership
fence: store it in HeldLock, pass it into the spawned guard task, and require
matching IDs for renewal, release, local map removal, and metadata deletion in
the lock lifecycle. In
gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/reaper.rs lines
174-195, restrict expiration reclamation to holder IDs owned by the local
instance. Add tests covering stale guards after reacquisition and races between
two reaper instances.
- Around line 329-340: Update try_acquire() and lock() so the entire pool
checkout and SQL acquisition path is bounded by timeout/deadline, rather than
checking the deadline only between attempts. Ensure cancellation or any exit
after pg_try_advisory_lock succeeds unlocks the session or discards the
connection before it returns to the pool. Preserve the existing unavailable-lock
result while guaranteeing no pooled connection retains a live advisory lock.
In
`@gears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/lock/0002_cluster_lock.sql`:
- Around line 17-23: Update the cluster_lock migration DDL to create the table
in the schema specified by config.schema, ensuring it matches the schema used by
runtime lock queries. Qualify the CREATE TABLE target or explicitly set the
migration search path, and reject unsupported schema values if required;
preserve the existing table definition and primary key.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/plugin.rs`:
- Around line 136-164: Ensure the LISTEN readiness step via spawn_listen_task
completes successfully before starting cache_reaper and lock_reaper, or
explicitly cancel and await both reapers when it fails. Update the
build_and_start flow around spawn_cache_reaper, spawn_reaper, and
spawn_listen_task so a startup error cannot leave reapers detached and holding
the pool alive.
---
Outside diff comments:
In `@gears/system/cluster/cluster-conformance/src/cache.rs`:
- Around line 321-329: Bound the asynchronous waits in cache.rs: at lines
321-329, wrap the expiry-event wait around wait_for and at lines 433-473, wrap
both Changed and Deleted waits with the existing timeout mechanism. Ensure
timeout failures become assertion failures rather than allowing the conformance
suite to hang, while preserving the current event predicates and success
behavior.
---
Minor comments:
In `@gears/system/cluster/cluster-conformance/src/lock.rs`:
- Around line 145-150: Adjust the timing in the SC-LOCK-005 renewal test so the
lease is renewed with a larger safety margin before expiry: reduce the elapsed
duration in the time.elapse call preceding guard.renew from 150 ms to 50 ms,
while preserving the 200 ms lease and existing success assertion.
In `@gears/system/cluster/cluster-conformance/src/time.rs`:
- Around line 65-72: Update the documentation for Time::elapse to state that
Virtual advances by dur, while Real sleeps for the capped duration
dur.min(Self::REAL_CAP), so the Real variant does not promise that the full
requested duration elapses.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.md`:
- Around line 59-60: Update the cache/watch.rs test descriptions to use the
implemented 7997-byte key limit instead of 8190, including the corresponding
occurrences at the additional referenced sections. Keep the descriptions aligned
with MAX_KEY_BYTES and DESIGN.md so the documented boundary coverage remains
consistent.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rs`:
- Around line 179-195: Update both delete SQL statements in the `delete`
implementations to include the same live-entry expiration predicate used by
`get`, `contains`, and CAS, so expired rows are neither removed nor reported as
deleted. Preserve the existing transaction, notification, and boolean-result
behavior for live rows.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs`:
- Around line 113-145: Remove the claim that PostgreSQL queue overflow produces
an empty payload while preserving ParsedNotification::Reset for empty,
malformed, and external payloads. In
gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs lines
113-145, update the parse_notification documentation; in docs/DESIGN.md lines
123 and 388-390, describe overflow as a commit failure and clarify queue usage;
in docs/TESTING.md lines 259-260 and 279, retitle the cases for explicit
empty-payload handling or commit-failure behavior.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/config.rs`:
- Around line 101-112: Validate cache_reaper_interval_ms and
lock_reaper_interval_ms as non-zero during configuration deserialization or
startup before they reach tokio::time::interval. Update the configuration
validation associated with these fields, preserving their existing defaults and
rejecting zero values with a clear configuration error.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/tests/common/mod.rs`:
- Around line 290-303: Update wait_until so each condition().await is bounded by
the overall deadline, preventing stalled SQL or network I/O from exceeding
timeout indefinitely. Compute the remaining duration before invoking condition
and apply it to the await (for example, with Tokio’s timeout); return false when
the condition exceeds the deadline, while preserving the existing success and
polling behavior.
In
`@gears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rs`:
- Around line 101-125: The integration test must synchronize with the waiter’s
registration before calling guard.release(), because waiter.is_finished() only
confirms the task is running, not that lock() reached ReleaseWaiters. Update the
waiter setup around lock() to expose a test-only registration barrier or
observable waiter count, await that confirmation, then release the guard so the
latency assertion specifically exercises the NOTIFY wake path.
In
`@gears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rs`:
- Around line 46-48: Update the timeout wrapping watch.recv() in the event
delivery test to use the standard watch-delivery timeout established by the
neighboring LISTEN/NOTIFY tests, replacing the 500 ms deadline while preserving
the existing timeout failure assertion.
---
Nitpick comments:
In `@gears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rs`:
- Around line 20-23: Update the module overview near lock_conformance to
describe the implemented isolation: scenarios share one container and use
per-schema tables plus drain_held, rather than receiving a fresh container per
scenario. Keep the existing search_path and server-wide advisory-lock context
accurate.
🪄 Autofix (Beta)
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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 4c376a9c-2f38-40fc-a42e-ae2a35640028
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (45)
Cargo.tomlgears/system/cluster/cluster-conformance/src/cache.rsgears/system/cluster/cluster-conformance/src/discovery.rsgears/system/cluster/cluster-conformance/src/factory.rsgears/system/cluster/cluster-conformance/src/leader.rsgears/system/cluster/cluster-conformance/src/lib.rsgears/system/cluster/cluster-conformance/src/lock.rsgears/system/cluster/cluster-conformance/src/time.rsgears/system/cluster/cluster-conformance/tests/self_test.rsgears/system/cluster/cluster/Cargo.tomlgears/system/cluster/cluster/src/defaults/discovery.rsgears/system/cluster/cluster/src/defaults/discovery_tests.rsgears/system/cluster/cluster/src/defaults/test_cache.rsgears/system/cluster/cluster/src/gear.rsgears/system/cluster/cluster/src/wiring.rsgears/system/cluster/cluster/tests/conformance.rsgears/system/cluster/plugins/postgres-cluster-plugin/Cargo.tomlgears/system/cluster/plugins/postgres-cluster-plugin/README.mdgears/system/cluster/plugins/postgres-cluster-plugin/docs/DESIGN.mdgears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.mdgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/reaper.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/config.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lib.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/lock_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/notify.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/reaper.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/cache/0001_cluster_cache.sqlgears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/lock/0002_cluster_lock.sqlgears/system/cluster/plugins/postgres-cluster-plugin/src/pg_error.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/pg_setup.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/plugin.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/provider.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/cache_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/common/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/lifecycle_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/watch_integration.rstools/dylint_lints/de07_security/de0706_no_direct_sqlx/src/lib.rstools/dylint_lints/lint_utils/src/lib.rs
💤 Files with no reviewable changes (1)
- gears/system/cluster/cluster/src/defaults/test_cache.rs
asmith987
force-pushed
the
feat/cluster-postgres
branch
from
July 16, 2026 21:45
5d5a4d4 to
cc4c4e8
Compare
Contributor
Author
Review follow-up — remaining (Minor / Nitpick) itemsThe Critical/Major inline comments each have a threaded reply above. The Minor/Nitpick items posted in the review summary (no inline thread to reply to) are addressed as follows:
Not applied (false positives): the Verification: |
There was a problem hiding this comment.
Actionable comments posted: 7
🤖 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 `@gears/system/cluster/cluster/src/defaults/discovery.rs`:
- Around line 484-505: Update the local pre-insert path in register, where the
registry map is modified, to call Registry::bump_generation for the affected
service whenever the insertion changes the map. Ensure this occurs before an
in-flight reconciliation can commit, while leaving the existing
StoreMaintainer::apply generation handling unchanged.
- Around line 675-688: Update the discovery sweep around cache.get and the fresh
view replacement so any backend get failure aborts the sweep instead of silently
skipping that key. Preserve the existing service view when a read fails, or
propagate the error through the surrounding discovery flow; only replace the
full view after every key has been read successfully.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/docs/DESIGN.md`:
- Line 70: Update the table-count statement in the PostgreSQL cluster plugin
design documentation from three to two, while retaining the existing schema and
migration details; do not count the virtual cluster_lock_notify NOTIFY channel
as a database table.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.md`:
- Around line 79-97: The conformance factories make_cache and make_lock must
retain their plugin handles instead of returning only cloned backends. Introduce
or reuse an owning scenario fixture that stores each handle and exposes async
teardown via stop(), then update the leader and discovery factories around the
referenced sections to follow the same pattern. Ensure the runner invokes stop()
after each scenario so handles are not dropped prematurely and background tasks
are cleanly terminated.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/mod.rs`:
- Around line 350-359: Update try_acquire and the corresponding release path to
reject lock names whose UTF-8 byte length exceeds PostgreSQL’s 7999-byte NOTIFY
payload limit before inserting or acquiring the lock. Return the existing
validation/error type without mutating lock state, and add boundary tests
covering 7999-byte acceptance and oversized-name rejection.
In `@tools/dylint_lints/lint_utils/src/lib.rs`:
- Around line 253-267: Update is_in_postgres_cluster_plugin_path and its
path-matching helper check_span_path to use component-aware matching or explicit
separator boundaries rather than substring contains. Preserve the supported
absolute, relative, and simulated_dir paths while rejecting prefixed unrelated
paths such as not-gears or my-postgres-cluster-plugin.
- Around line 253-267: Add tests covering absolute, relative, and simulated-dir
matches and near-miss exclusions for is_in_postgres_cluster_plugin_path. Also
implement the missing test_comment_annotations_match_stderr in lint_utils,
reusing the existing annotation/stderr comparison patterns and test helpers.
🪄 Autofix (Beta)
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: defaults
Review profile: CHILL
Plan: Pro
Run ID: ecb49d3c-7a07-4a48-a0c4-b896ad010331
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (47)
Cargo.tomlgears/system/cluster/cluster-conformance/Cargo.tomlgears/system/cluster/cluster-conformance/src/cache.rsgears/system/cluster/cluster-conformance/src/discovery.rsgears/system/cluster/cluster-conformance/src/factory.rsgears/system/cluster/cluster-conformance/src/leader.rsgears/system/cluster/cluster-conformance/src/lib.rsgears/system/cluster/cluster-conformance/src/lock.rsgears/system/cluster/cluster-conformance/src/time.rsgears/system/cluster/cluster-conformance/tests/self_test.rsgears/system/cluster/cluster/Cargo.tomlgears/system/cluster/cluster/src/defaults/discovery.rsgears/system/cluster/cluster/src/defaults/discovery_tests.rsgears/system/cluster/cluster/src/defaults/test_cache.rsgears/system/cluster/cluster/src/gear.rsgears/system/cluster/cluster/src/wiring.rsgears/system/cluster/cluster/tests/conformance.rsgears/system/cluster/plugins/postgres-cluster-plugin/Cargo.tomlgears/system/cluster/plugins/postgres-cluster-plugin/README.mdgears/system/cluster/plugins/postgres-cluster-plugin/docs/DESIGN.mdgears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.mdgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/reaper.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/config.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/config_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lib.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/lock_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/notify.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/reaper.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/cache/0001_cluster_cache.sqlgears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/lock/0002_cluster_lock.sqlgears/system/cluster/plugins/postgres-cluster-plugin/src/pg_error.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/pg_setup.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/plugin.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/provider.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/cache_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/common/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/lifecycle_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/watch_integration.rstools/dylint_lints/de07_security/de0706_no_direct_sqlx/src/lib.rstools/dylint_lints/lint_utils/src/lib.rs
💤 Files with no reviewable changes (1)
- gears/system/cluster/cluster/src/defaults/test_cache.rs
🚧 Files skipped from review as they are similar to previous changes (29)
- gears/system/cluster/plugins/postgres-cluster-plugin/README.md
- gears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/lock/0002_cluster_lock.sql
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lib.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/pg_error.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/cache/0001_cluster_cache.sql
- gears/system/cluster/cluster/Cargo.toml
- Cargo.toml
- gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/reaper.rs
- gears/system/cluster/cluster-conformance/tests/self_test.rs
- tools/dylint_lints/de07_security/de0706_no_direct_sqlx/src/lib.rs
- gears/system/cluster/cluster/src/gear.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/lock_tests.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/watch_integration.rs
- gears/system/cluster/cluster-conformance/src/discovery.rs
- gears/system/cluster/cluster-conformance/src/factory.rs
- gears/system/cluster/cluster-conformance/src/time.rs
- gears/system/cluster/cluster/src/defaults/discovery_tests.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rs
- gears/system/cluster/cluster/tests/conformance.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/lifecycle_integration.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/notify.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/provider.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/plugin.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/cache_integration.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/reaper.rs
- gears/system/cluster/cluster-conformance/src/leader.rs
asmith987
force-pushed
the
feat/cluster-postgres
branch
3 times, most recently
from
July 17, 2026 01:14
09c96a4 to
5e88ed4
Compare
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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 `@gears/system/cluster/plugins/postgres-cluster-plugin/docs/DESIGN.md`:
- Around line 340-345: Update the SQL contract entries for delete and
compare_and_delete to exclude expired rows, matching the implementation’s
active-entry predicates. Update scan_prefix to document the escaped bound
pattern and its ESCAPE '\' clause instead of appending a literal '%' in SQL,
while preserving the documented prefix behavior.
- Around line 450-453: Update the PgBouncer incompatibility explanation near the
documented startup validation to state that returning a connection in
transaction pooling can leave the session advisory lock attached to the pooled
server session and leak across clients, rather than claiming the lock is
released between transactions. Preserve the existing build_and_start rejection
and operator guidance.
- Line 123: Update the DESIGN.md descriptions of NOTIFY queue overflow,
including the referenced sections, to state that overflow rejects the producer
transaction without inherently disconnecting the listener or incrementing
cluster_watch_resets_total. Reserve reconnect/reset terminology for actual
LISTEN connection gaps, and direct overflow monitoring to write/provider errors
and PostgreSQL logs.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/mod.rs`:
- Around line 329-340: Update reassert_synchronous_commit and its callers to
accept the guard’s holder_id, then verify the current entry for the name still
belongs to that holder before executing SQL; return without reasserting on
mismatch so stale guards cannot affect successor holders. Apply the same
holder_id fencing in the periodic guard path around lines 543-579 and update
__test_reassert_synchronous_commit and its test callers to pass the holder
identifier.
In `@gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/notify.rs`:
- Around line 206-250: The initial connection in spawn_release_listen_task must
use reconnect_with_backoff instead of returning immediately after
connect_and_listen fails. Rework the setup so the first attempt and subsequent
retries share the cancellation-aware retry loop, and ensure reconnect attempts
can be interrupted by cancel while awaiting connect_and_listen rather than
blocking shutdown.
🪄 Autofix (Beta)
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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 8a9046cd-a9df-4630-9cae-74dd129cdac2
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (47)
Cargo.tomlgears/system/cluster/cluster-conformance/Cargo.tomlgears/system/cluster/cluster-conformance/src/cache.rsgears/system/cluster/cluster-conformance/src/discovery.rsgears/system/cluster/cluster-conformance/src/factory.rsgears/system/cluster/cluster-conformance/src/leader.rsgears/system/cluster/cluster-conformance/src/lib.rsgears/system/cluster/cluster-conformance/src/lock.rsgears/system/cluster/cluster-conformance/src/time.rsgears/system/cluster/cluster-conformance/tests/self_test.rsgears/system/cluster/cluster/Cargo.tomlgears/system/cluster/cluster/src/defaults/discovery.rsgears/system/cluster/cluster/src/defaults/discovery_tests.rsgears/system/cluster/cluster/src/defaults/test_cache.rsgears/system/cluster/cluster/src/gear.rsgears/system/cluster/cluster/src/wiring.rsgears/system/cluster/cluster/tests/conformance.rsgears/system/cluster/plugins/postgres-cluster-plugin/Cargo.tomlgears/system/cluster/plugins/postgres-cluster-plugin/README.mdgears/system/cluster/plugins/postgres-cluster-plugin/docs/DESIGN.mdgears/system/cluster/plugins/postgres-cluster-plugin/docs/TESTING.mdgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/reaper.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/config.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/config_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lib.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/lock_tests.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/notify.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/lock/reaper.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/cache/0001_cluster_cache.sqlgears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/lock/0002_cluster_lock.sqlgears/system/cluster/plugins/postgres-cluster-plugin/src/pg_error.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/pg_setup.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/plugin.rsgears/system/cluster/plugins/postgres-cluster-plugin/src/provider.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/cache_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/common/mod.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/lifecycle_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rsgears/system/cluster/plugins/postgres-cluster-plugin/tests/watch_integration.rstools/dylint_lints/de07_security/de0706_no_direct_sqlx/src/lib.rstools/dylint_lints/lint_utils/src/lib.rs
💤 Files with no reviewable changes (1)
- gears/system/cluster/cluster/src/defaults/test_cache.rs
🚧 Files skipped from review as they are similar to previous changes (36)
- Cargo.toml
- gears/system/cluster/plugins/postgres-cluster-plugin/README.md
- gears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/cache/0001_cluster_cache.sql
- gears/system/cluster/cluster-conformance/src/lib.rs
- gears/system/cluster/cluster-conformance/tests/self_test.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/migrations/lock/0002_cluster_lock.sql
- gears/system/cluster/cluster-conformance/src/factory.rs
- gears/system/cluster/cluster/Cargo.toml
- gears/system/cluster/cluster/tests/conformance.rs
- gears/system/cluster/cluster-conformance/Cargo.toml
- gears/system/cluster/cluster/src/gear.rs
- gears/system/cluster/cluster-conformance/src/time.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/provider.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/watch_integration.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/config_tests.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/reaper.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lib.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/conformance.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/config.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/plugin.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/Cargo.toml
- gears/system/cluster/cluster-conformance/src/discovery.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/postgres_specific.rs
- gears/system/cluster/cluster/src/wiring.rs
- tools/dylint_lints/de07_security/de0706_no_direct_sqlx/src/lib.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/lock_integration.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/reaper.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/lock/lock_tests.rs
- gears/system/cluster/cluster-conformance/src/leader.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch_tests.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/src/cache/watch.rs
- gears/system/cluster/cluster/src/defaults/discovery.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/cache_integration.rs
- gears/system/cluster/cluster-conformance/src/cache.rs
- gears/system/cluster/plugins/postgres-cluster-plugin/tests/lifecycle_integration.rs
- gears/system/cluster/cluster-conformance/src/lock.rs
Adds the Postgres cache and distributed-lock provider implementations for the cluster gear (sqlx-backed pool, pg_advisory_lock, LISTEN/NOTIFY watch), wires both providers into the cluster gear's provider registry, and extends the shared conformance suite so it runs against the new backend alongside the existing standalone plugin. Signed-off-by: Andre Smith <andre.smith+oss@acronis.com>
asmith987
force-pushed
the
feat/cluster-postgres
branch
from
July 17, 2026 02:38
f1560c4 to
54106de
Compare
| Err(_lost) => { | ||
| match reconnect_with_backoff(&connection_string, &policy, &mut attempt, &cancel).await { | ||
| Some(reconnected) => listener = reconnected, | ||
| None => return, |
Collaborator
There was a problem hiding this comment.
MEDIUM
On reconnect-budget exhaustion the release-listen background task returns silently, conflating a terminal connection-loss failure with graceful cancellation and emitting no warn/error log or consumer signal.
Distinguish cancellation from budget exhaustion and log at warn/error (and/or signal waiters) on exhaustion before the task exits, so the permanent loss of NOTIFY wakeups is observable.
| () = cancel.cancelled() => return None, | ||
| () = tokio::time::sleep(backoff) => {} | ||
| } | ||
| if let Ok(listener) = connect_and_listen(connection_string).await { |
Collaborator
There was a problem hiding this comment.
MEDIUM
The LISTEN task's mid-stream reconnect awaits connect_and_listen directly without selecting on the CancellationToken, so a shutdown fired while PgListener::connect reaches an unreachable host stalls stop() for the full connect timeout.
Wrap the reconnect connect in a cancellation-aware helper as lock/notify.rs does (connect_and_listen_cancellable selecting on cancel.cancelled()), returning None on cancellation so the task exits promptly.
| // `PgListener`'s own transparent reconnect just ran and | ||
| // succeeded; events during the gap may have been missed. | ||
| attempt = 0; | ||
| registry.dispatch(&ParsedNotification::Reset).await; |
Collaborator
There was a problem hiding this comment.
MEDIUM
DESIGN §8 says the LISTEN task's Reset broadcasts call ClusterMetrics::watch_reset("cache") (backing cluster_watch_resets_total) and emit a cluster.watch.reset WARN log, but the watch task takes no ClusterMetrics and emits neither signal on any Reset.
Thread the ClusterMetrics sink into the LISTEN task/WatchRegistry and emit watch_reset("cache") plus a cluster.watch.reset WARN wherever a Reset is dispatched, or amend DESIGN §8 to state these are covered solely by the SDK combinator.
| /// operations emit the contracted `cluster.cache.*` signals (DESIGN.md §8), | ||
| /// mirroring `StandaloneClusterHandle`'s construction | ||
| /// (`standalone-cluster-plugin/src/plugin.rs`). | ||
| pub fn instrumented_cache( |
Collaborator
There was a problem hiding this comment.
LOW
instrumented_cache is declared pub but is called only once within plugin.rs, is not re-exported by lib.rs, and exposes the internal PostgresCache type in its signature.
Drop pub (or use pub(crate) only if cross-module reuse is intended) so the function's visibility matches its actual use.
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn dispatch_prunes_a_watcher_whose_consumer_dropped_it() { |
Collaborator
There was a problem hiding this comment.
LOW
The test named for pruning a dropped watcher makes no assertion and relies only on the absence of a panic, so it would pass even if the dead entry were never pruned.
Assert the registry actually removed the dropped watcher's entry after dispatch (inspect the internal subscriber map as the notify.rs tests do).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds the Postgres cache and distributed-lock provider implementations for the cluster gear (sqlx-backed pool, pg_advisory_lock, LISTEN/NOTIFY watch), wires both providers into the cluster gear's provider registry, and extends the shared conformance suite so it runs against the new backend alongside the existing standalone plugin.
Summary by CodeRabbit