feat(lineage): register_per_org_sweep seam for per-org reclamation - #285
Conversation
…vernance start-gate
…d (preserve governance start invariant) + drop stale capability refs
|
Warning Review limit reached
Next review available in: 47 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: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughAdds a per-org sweep hook registry ( ChangesPer-org sweep hook mechanism
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Scheduler
participant GcTick as _gc_tick
participant SweepHook as Registered per-org sweep
participant Anomaly as capture_anomaly
Scheduler->>GcTick: run tick for org
GcTick->>SweepHook: invoke(org_id, now)
alt sweep succeeds
SweepHook-->>GcTick: return
else sweep raises
SweepHook-->>GcTick: exception
GcTick->>Anomaly: capture_anomaly(exception)
GcTick->>GcTick: continue with next sweep
end
Possibly related PRs
🚥 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.
🧹 Nitpick comments (3)
reflexio/server/services/lineage/gc_scheduler.py (3)
366-408: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
maybe_start_lineage_gc's docstring is now out of sync with its widened start gate.The docstring (lines 372-377) still states: "Startup runs when bootstrap-org config sets EITHER
lineage_gc.enabled... ORexpiry_reclamation.enabled... ReturnsNoneif neither flag is set." That's no longer accurate — the actual gate at lines 448-454 also starts the scheduler whengovernance_retention.audit_events_retention_enabledis set, or when any global/per-org sweep hook is registered, regardless of the two documented flags. This is a meaningful behavioral surface (it's precisely the widening this PR introduces per the PR objectives) and leaving the docstring stale could mislead future readers/operators about when the scheduler actually starts.📝 Suggested docstring update
"""Start the scheduler when bootstrap config enables tombstone GC or expiry reclamation. - Startup runs when bootstrap-org config sets EITHER ``lineage_gc.enabled`` - (Class A: profile expiry + tombstone GC) OR ``expiry_reclamation.enabled`` - (Class B: plain-row direct-delete sweeps). Returns ``None`` if neither flag - is set. + Startup runs when bootstrap-org config sets ``lineage_gc.enabled`` (Class A), + ``expiry_reclamation.enabled`` (Class B), or + ``governance_retention.audit_events_retention_enabled``, OR when any sweep + hook has been registered via :func:`register_per_org_sweep` or + :func:`register_global_sweep` (registered hooks carry their own per-org + gate in the enterprise closure). Returns ``None`` only if none of these + hold.Also applies to: 440-454
🤖 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 `@reflexio/server/services/lineage/gc_scheduler.py` around lines 366 - 408, Update the docstring for maybe_start_lineage_gc so it matches the full startup gate now implemented in the function: describe that the scheduler can start not only from lineage_gc.enabled or expiry_reclamation.enabled, but also when governance_retention.audit_events_retention_enabled is set or when a global/per-org sweep hook is registered. Keep the return description aligned with the actual behavior and ensure the wording in the docstring near the startup criteria reflects the checks performed in maybe_start_lineage_gc rather than only the older two-flag path.
102-131: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueRegistry mutation isn't synchronized with the tick-loop reader.
_per_org_sweep_hooksis a plain module-level list, appended/cleared fromregister_per_org_sweep/clear_per_org_sweepswhile_gc_tickiterates it on the scheduler thread. This mirrors the pre-existing_global_sweep_hookspattern, so the incremental risk here is low (hooks are meant to be registered once at startup and cleared only in tests), but aclear()racing with an in-progress iteration on a live scheduler thread could still surface subtleRuntimeError/skipped-item behavior in CPython.🤖 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 `@reflexio/server/services/lineage/gc_scheduler.py` around lines 102 - 131, The per-org sweep registry in gc_scheduler is mutated unsafely while _gc_tick may be iterating it, so make register_per_org_sweep and clear_per_org_sweeps safe against concurrent scheduler-thread reads. Update the _per_org_sweep_hooks handling to use the same safe iteration/mutation approach as the existing _global_sweep_hooks pattern, or otherwise snapshot/protect access inside _gc_tick so a clear() during iteration cannot cause skipped hooks or RuntimeError. Refer to _per_org_sweep_hooks, register_per_org_sweep, clear_per_org_sweeps, and _gc_tick when applying the fix.
309-327: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePer-sweep timestamp drift and discarded return value.
Two small inconsistencies versus the sibling sweep classes in this same function:
int(time.time())is recomputed for every hook inside the loop (line 314), whereas Class B computesnowonce before its loop (line 283) and reuses it across sweeps in the same tick.- The sweep's return value (documented as "the number of rows deleted") is discarded entirely — Class B (289-296) and
_run_global_sweeps(344-346) both log non-zero deletions for observability, but per-org sweeps get no equivalent visibility here.Neither is a functional bug (isolation/anomaly capture is correct), but aligning with the existing patterns would improve consistency and ops visibility into this enterprise-only reclamation path.
♻️ Optional consistency fix
# Per-org sweeps: invoked unconditionally (not gated on lineage_gc or # expiry_reclamation — the real gate lives in each enterprise closure). # Each sweep is isolated so one failure does not skip the rest. + now = int(time.time()) for sweep in _per_org_sweep_hooks: try: - sweep(org_id, int(time.time())) + deleted = sweep(org_id, now) + if deleted: + logger.info( + "event=per_org_sweep org_id=%s sweep=%s deleted=%d", + org_id, + getattr(sweep, "__qualname__", repr(sweep)), + deleted, + ) except Exception:🤖 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 `@reflexio/server/services/lineage/gc_scheduler.py` around lines 309 - 327, The per-org sweep loop in `gc_scheduler.py` should match the existing sweep patterns by computing the current timestamp once before iterating `_per_org_sweep_hooks` and reusing it for each `sweep(org_id, now)` call. Also capture each sweep’s return value and, when it indicates rows were deleted, emit the same kind of observability log used by the sibling class-B sweep path and `_run_global_sweeps`. Keep the existing isolation and exception handling in place, and update the per-org sweep block around `capture_anomaly`/`logger.exception` without changing the enterprise gate behavior.
🤖 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.
Nitpick comments:
In `@reflexio/server/services/lineage/gc_scheduler.py`:
- Around line 366-408: Update the docstring for maybe_start_lineage_gc so it
matches the full startup gate now implemented in the function: describe that the
scheduler can start not only from lineage_gc.enabled or
expiry_reclamation.enabled, but also when
governance_retention.audit_events_retention_enabled is set or when a
global/per-org sweep hook is registered. Keep the return description aligned
with the actual behavior and ensure the wording in the docstring near the
startup criteria reflects the checks performed in maybe_start_lineage_gc rather
than only the older two-flag path.
- Around line 102-131: The per-org sweep registry in gc_scheduler is mutated
unsafely while _gc_tick may be iterating it, so make register_per_org_sweep and
clear_per_org_sweeps safe against concurrent scheduler-thread reads. Update the
_per_org_sweep_hooks handling to use the same safe iteration/mutation approach
as the existing _global_sweep_hooks pattern, or otherwise snapshot/protect
access inside _gc_tick so a clear() during iteration cannot cause skipped hooks
or RuntimeError. Refer to _per_org_sweep_hooks, register_per_org_sweep,
clear_per_org_sweeps, and _gc_tick when applying the fix.
- Around line 309-327: The per-org sweep loop in `gc_scheduler.py` should match
the existing sweep patterns by computing the current timestamp once before
iterating `_per_org_sweep_hooks` and reusing it for each `sweep(org_id, now)`
call. Also capture each sweep’s return value and, when it indicates rows were
deleted, emit the same kind of observability log used by the sibling class-B
sweep path and `_run_global_sweeps`. Keep the existing isolation and exception
handling in place, and update the per-org sweep block around
`capture_anomaly`/`logger.exception` without changing the enterprise gate
behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 28e69e6e-aa98-4fcc-9510-e902661d949b
📒 Files selected for processing (3)
reflexio/models/config_schema.pyreflexio/server/services/lineage/gc_scheduler.pytests/server/services/lineage/test_gc_scheduler.py
…rg deleted count, snapshot hook lists, sync maybe_start docstring
b734604 to
ae28e19
Compare
|
Addressed all 3 nitpicks in
@coderabbitai review |
🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
cd /tmp
gh pr view 285 --repo ReflexioAI/reflexio --json commits --jq '.commits[-3:]' || true
gh pr diff 285 --repo ReflexioAI/reflexio > /tmp/pr285.diff
wc -l /tmp/pr285.diffLength of output: 1553 🏁 Script executed: #!/bin/bash
set -euo pipefail
fd gc_scheduler.py
cat -n reflexio/server/services/lineage/gc_scheduler.py | sed -n '1,260p'Length of output: 13628 🏁 Script executed: #!/bin/bash
set -euo pipefail
sed -n '260,420p' reflexio/server/services/lineage/gc_scheduler.pyLength of output: 7827 🏁 Script executed: #!/bin/bash
set -euo pipefail
sed -n '420,470p' reflexio/server/services/lineage/gc_scheduler.pyLength of output: 2641 Everything checks out. All three nitpicks are properly addressed in
Looks good to me overall. ✅ Action performedReview finished.
|
1 participant
Summary
Adds a per-org reclamation sweep seam to the lineage GC scheduler — the per-org counterpart to the merged
register_global_sweep(global) seam (#283). A deployment can register aCallable[[str, int], int]that the scheduler invokes once per org inside its existing per-org tick loop, keeping the OSS scheduler config-agnostic (no enterprise imports).The enterprise side uses this to fold its per-org
GovernanceRetentionScheduler(audit-event retention) into the shared scheduler and delete a standalone daemon (see the enterprise PR).Changes
gc_scheduler.py:_per_org_sweep_hookslist +register_per_org_sweep(fn: Callable[[str, int], int])+clear_per_org_sweeps()(test hook), mirroringregister_global_sweep.LineageGCScheduler._gc_tick: after the Class-B block, inside the per-org loop, invokes each registered per-org sweep unconditionally with(org_id, now)and per-sweep failure isolation (capture_anomaly("lineage.per_org_sweep.failed", org_id=, sweep=)). The gate is the enterprise closure's own per-org config — notlineage_gc/expiry_reclamation.maybe_start_lineage_gc: start-gate widened to also start when any reclamation sweep is registered (_per_org_sweep_hooks/_global_sweep_hooksnon-empty), so a per-org reclaimer whose real gate lives in a non-bootstrap tenant's config is never silently skipped (preserving the folded daemon's unconditional-start invariant).GovernanceRetentionCapabilityreferences replaced with the generic per-org-sweep description.Tests
test_gc_scheduler.py: per-org sweep runs once per org; failure-isolation (anomaly + siblings continue);maybe_startstarts when a per-org sweep is registered even with all config flags off; existingtest_gc_tick_never_runs_governance_retentionstays green (empty hooks → OSS never runs governance). 46 lineage tests pass.With no registered sweeps (OSS default), behavior is byte-for-byte identical to before.
Related PRs
Summary by CodeRabbit
New Features
Bug Fixes