Repository navigation
fix(events): preserve thread seq watermark across run deletion - #6477
Conversation
github-actions
Bot
added
area:backend
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Reviewed at cf59901. The core fix is sound: _max_seq_for_thread now takes max(row_max, watermark) under the existing thread mutation fence, all three write paths raise the watermark to every allocated seq (so any later partial deletion can't lower the floor), delete_by_run persists the watermark in the same transaction as the delete (crash-atomic), and migration 0034 backfills from existing rows. TDD coverage is good — delete-then-append with a strictly-increasing-seq assertion runs for all three backends and all three DB write paths, including the after_seq pagination symptom from #6471. Two non-blocking issues: the new migration test fails ruff format --check (backend CI enforces it), and the watermark-cleanup branch in delete_by_thread is unreachable from the production Gateway thread-deletion flow, so watermark rows for deleted threads are never garbage-collected.
| "content": "", | ||
| "event_metadata": {}, | ||
| "seq": 1, | ||
| "created_at": datetime.now(UTC), |
There was a problem hiding this comment.
The three "created_at" entries here and at lines 92 and 103 are over-indented by 4 spaces relative to their sibling keys. Backend CI enforces ruff format --check, and with the repo's backend/ruff.toml this file is rejected:
$ ruff format --check tests/test_migration_0034_run_event_seq_watermark.py
Would reformat: tests/test_migration_0034_run_event_seq_watermark.py
(ruff wants each key at the same 24-space indent as "seq".) Please run make format (or ruff format on this file) before pushing, otherwise CI will block the PR.
There was a problem hiding this comment.
Addressed in bc25cb5: the migration test is formatted and ruff format --check now passes.
| count = await session.scalar(count_stmt) or 0 | ||
| if count > 0: | ||
| await session.execute(delete(RunEventRow).where(*count_conditions)) | ||
| if resolved_user_id is None: |
There was a problem hiding this comment.
This cleanup branch is unreachable from the production thread-deletion flow. The Gateway deletes threads with the authenticated owner's scope (backend/app/gateway/routers/threads.py _delete_thread_data_with_reservation calls delete_events(thread_id, user_id=get_effective_user_id()), which returns a non-None str), and user_id=None is documented in resolve_user_id as reserved for migration scripts and CLI tools. So in practice every user-driven thread delete orphans its run_event_thread_seq row: the table grows by one row per deleted thread with no GC path, and the PR's stated "recreated thread restarts at seq 1, matching memory and JSONL" behavior never actually triggers on the main path. This branch also has no test coverage.
Suggested fix that still preserves the anti-reuse invariant when other users' rows survive the scoped delete: drop the watermark whenever no event rows remain for the thread, e.g. run unconditionally in place of the resolved_user_id is None guard:
DELETE FROM run_event_thread_seq
WHERE thread_id = :tid
AND NOT EXISTS (SELECT 1 FROM run_events WHERE thread_id = :tid)That covers the owner-scoped HTTP path (the common single-owner case), keeps the floor when a shared thread still has other users' events, and leaves the CLI full wipe behaving as before.
There was a problem hiding this comment.
Addressed in bc25cb5. Cleanup now runs unconditionally with a NOT EXISTS check on the thread, so owner-scoped and unscoped deletion collect the watermark only when the thread is empty while surviving events retain the allocation floor. Tests cover owner-scoped/unscoped deletion, already-empty threads, remaining owners/categories, recreation at seq 1, and cursor visibility. All PR checks are green.
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Re-reviewed at head bc25cb5. Both previously flagged items are resolved: the migration test now passes ruff format --check, and the watermark cleanup in delete_by_thread runs unconditionally behind the NOT EXISTS guard, so the owner-scoped Gateway path now garbage-collects the watermark for emptied threads (covered by the new test_delete_by_thread_clears_watermark_when_empty and test_owner_scoped_thread_delete_preserves_watermark_for_remaining_events cases). I verified at head: test_run_event_store.py + test_db_event_store_lock_lifecycle.py + test_migration_0034_run_event_seq_watermark.py give 119 passed / 1 skipped, and ruff format/check are clean on the touched files. All three RunEventRow creation sites raise the watermark, so no allocation path bypasses the floor. One remaining nit below.
| async with self._sf() as session: | ||
| async with session.begin(): | ||
| await self._acquire_thread_mutation_fence(session, thread_id) | ||
| max_seq = await self._max_seq_for_thread(session, thread_id) |
There was a problem hiding this comment.
Nit (no functional issue): this takes the thread mutation fence a second time — delete_by_run already acquires it on the previous line, and _max_seq_for_thread() fences internally too. On Postgres that is a redundant pg_advisory_xact_lock round trip per run deletion (the lock is re-entrant within the transaction, so it cannot deadlock); on SQLite the fence is a no-op either way. Since the xact-scoped advisory lock persists for the rest of the transaction, the standalone _acquire_thread_mutation_fence() above could be dropped and the fencing left to _max_seq_for_thread(), keeping the count/delete below equally protected. The other _max_seq_for_thread() call sites (put, put_batch, put_if_absent) are correct as-is because they have no outer fence.
There was a problem hiding this comment.
Addressed in 6fc7fcf. delete_by_run now relies on the transaction-scoped fence acquired inside _max_seq_for_thread, so the count/delete remain under the same lock without a second Postgres advisory-lock round trip. The existing advisory-lock test now also asserts exactly one acquisition.
delete_by_run already reaches _max_seq_for_thread, which acquires the transaction-scoped thread mutation fence. Keep the count and delete under that same transaction lock without taking the Postgres advisory lock twice, and pin the single acquisition in the existing test.
…ermark test Upstream bytedance#6477 added test_migration_0034_run_event_seq_watermark.py asserting 0034 is the single head; this branch chains 0035_login_throttle after it, so the test now pins the current head and walks the chain like its siblings.
…as (#6501) * fix(auth): count login lockouts in the database across Gateway replicas POST /api/v1/auth/login/local throttled failed logins per client IP in a module-level dict, so with N Gateway replicas behind one load balancer an attacker got N x auth.local.max_login_attempts guesses and a lockout on one replica was invisible to the others. Move the counter behind a LoginThrottleStore (deerflow.persistence. login_throttle): MemoryLoginThrottleStore keeps the historical per-process semantics and 10k tracked-IP bound; SqlLoginThrottleStore shares one login_throttle row per IP through the application database, incrementing and starting the lock in a single dialect-native upsert, applying check decisions with compare-and-set predicates in a separate write transaction, and sweeping served locks plus 24h-idle counters in bounded batches. Both stores pin the existing contract: the lock starts when the count reaches max_login_attempts, the committed duration is honored when the policy shrinks mid-lock, a served sentence is never resurrected, tightening keeps the count, success resets the IP, and the 429 status and message are unchanged. New startup-only selector auth.local.throttle_storage (auto | memory | db, default auto = the database whenever database.backend is sqlite or postgres). The Gateway installs the store once at startup after the persistence engine; deps._validate_login_throttle_storage warns when a multi-process deployment keeps memory. Migration 0034_login_throttle creates the table; config_version 55 -> 56 in config.example.yaml and the Helm chart. Docs: README, CONFIGURATION.md, AUTH_DESIGN.md, config and migrations AGENTS.md, CHANGELOG. * docs(changelog): link the login throttle entry to #6501 * fix(auth): re-check expiry on the row the login-throttle sweep deletes The bounded cleanup in SqlLoginThrottleStore.record_failure deleted `WHERE ip IN (SELECT ip ... WHERE <served or stale> LIMIT 200)`. On PostgreSQL READ COMMITTED the expiry predicate applied only inside the snapshot-bound subquery: when the DELETE had to wait on a row lock held by a concurrent check() that was extending the same lock under a raised lockout_seconds, PostgreSQL re-evaluated only the outer WHERE on the updated tuple, IP membership still held for the old candidate, and the now-active lock was deleted; the next attempt from that IP got a clean record and reached password verification (review of #6501, P2). Repeat the served/stale predicate on the DELETE target as well (SqlLoginThrottleStore.sweep_statement), keeping the bounded candidate subquery, so the re-evaluation sees the updated row's expiry. The other writes already carry their full compare-and-set predicate on the target row: _discard and the check() UPDATE both use _matches (ip, fail_count, locked_at, lock_duration_seconds), so a tuple changed while they waited no longer matches and the stale decision is skipped. _discard keeps the equality compare-and-set rather than adding an expiry condition: its early-release branch deletes a lock whose committed sentence has not elapsed under the lowered live policy, which an expiry predicate would wrongly block, and the equality predicate already pins every input of the decision. reset stays unconditional by design. Tests: a dialect-compile test asserts the DELETE's own WHERE carries the expiry predicate for postgresql and sqlite (red on the previous HEAD), and a PostgreSQL-only test (skipped without TEST_POSTGRES_URI) reproduces the update-versus-sweep interleaving with two engines, waiting on pg_stat_activity for the sweep to block before the extender commits. Docs: sql.py concurrency contract and the migrations guide's 0034 bullet. * fix(auth): keep an active login lock's clock and fail closed on explicit db Address the "changes requested" review on #6501. Lock clock (confirmed defect): record_failure re-stamped locked_at = now for every failure on an already-locked row, so a sustained attacker's sentence slid forward with each attempt. Both stores now apply one rule in one atomic decision: new_count = fail_count + 1; below max_attempts the IP counts with no lock (a failure after the operator raised the threshold clears an existing lock); at or over it, a failure during an *active* lock keeps locked_at and the committed lock_duration unchanged, while a never-locked or served row starts anew with now / lockout_seconds (so the failure that reaches a raised threshold starts a fresh lock rather than keeping a stale locked_at). The SQL upsert expresses it as nested case() on the row's own values; the memory store mirrors it. The sweeps in both stores now leave the row being recorded to that decision instead of deleting a served lock or stale counter first, which split the contract (SQL would have produced a fresh (1, NULL, NULL) row). base.py states the rule. Explicit db downgrade: resolve_login_throttle_store raises SystemExit — the failure type of the other startup validations in app.gateway.deps — when throttle_storage='db' resolves to the shared table but no engine is initialised, so langgraph_runtime refuses to start instead of silently running every replica on per-process counters; the lazy bare-app path applies the same rule and installs nothing. 'auto' keeps warning and falling back. 'db' on a memory database still degrades with a warning (there is no shared table to insist on). Non-blocking items: _key() logs a warning when a client key is truncated to 255 characters (never the key itself); the sql.py docstring covers wall-clock locked_at across replicas (run NTP; skew shifts evaluation by that much) and why STALE_COUNTER_SECONDS=24h / SWEEP_BATCH_SIZE=200 were chosen; a store-level blocking-I/O anchor covers the SQL reset path; the _validate_login_throttle_storage warning was verified to render 'throttle_storage=memory' for the StrEnum (the gate test now passes the enum member and asserts that text). reset stays unconditional: a peer failure that commits after a reset leaves fail_count=1, one genuine failure after the cleared history, which a new test documents. Docs: CHANGELOG entry, config.example.yaml, CONFIGURATION.md, AUTH_DESIGN.md and the throttle_storage field description. * fix(auth): re-evaluate a login-throttle check whose compare-and-set missed SqlLoginThrottleStore.check applied its decision (commit the live duration, or clear a served / released lock) with a compare-and-set predicate on the snapshot it had read, but ignored the affected-row count. When a peer changed the row between the read and the write — an admitted wrong password incrementing fail_count on another replica, or a successful login clearing the row — the UPDATE or DELETE matched zero rows, yet check reported the decision as applied: it could return an extended 60s sentence while the row kept its 1s duration, so the next check cleared the lock and let password verification through; a live duration decrease could be dropped the same way; and a row a peer had just cleared was still reported as locked (#6501 review, P2). Every compare-and-set write in check now checks result.rowcount. On a miss the row is re-read and re-evaluated from the top, up to CHECK_CAS_ATTEMPTS (3) times, applying a fresh compare-and-set decision on the new snapshot and never clobbering a reset or replacement: a row that is gone is reported as allowed, a fresh peer lock is reported as locked, and an incremented count gets the extension or decrease committed on it. Exhausting the bound fails closed: nothing is written and the answer is the last snapshot's lock under the live policy, logged at debug. The read stays in its own short transaction and each write in a separate one. The evaluation is now one shared pure function, evaluate_check in base.py (CheckDecision), used by both stores so the contract cannot diverge. The memory store keeps a single pass: its compare-and-set cannot miss because there is no await between the read and the write on the loop thread, which its docstring now states. Tests (red first on df26c2a) use a session proxy that runs the peer mutation on a second engine right after the snapshot read ends its transaction: the reviewer's raised-duration interleaving (59.5s reported and committed, the increment survives, a later check still locked), the mirrored decrease, early release on the fresh row, a reset landing between read and write (0.0, no row recreated), and the bound-exhausted fail-closed case. * docs(changelog): resolve merge markers left in the Unreleased section * test(migrations): pin 0035_login_throttle as the head in the 0034 watermark test Upstream #6477 added test_migration_0034_run_event_seq_watermark.py asserting 0034 is the single head; this branch chains 0035_login_throttle after it, so the test now pins the current head and walks the chain like its siblings. * perf(auth): one session per login-throttle check, one transaction per failure Two non-blocking performance suggestions from the #6501 review. Single session per check. The router probed the row with store.get() and then called store.check(), which re-read it: a recorded IP cost two connection-pool checkouts and two SELECTs per request on the unauthenticated login path, letting a lockout flood compete with legitimate traffic for Postgres connections. LoginThrottleStore.check now takes the policy as an async callable (LoginThrottlePolicy) and resolves it lazily inside the store: the probe, the policy resolution and the compare-and-set loop share one session, a clean IP returns 0.0 without ever calling the policy (the config.yaml read is still skipped for the common case), and a recorded IP resolves it exactly once, also across compare-and-set retries. The decision is made on a snapshot read after the resolution: the policy read is a yield point, and a decision that needs no write has no compare-and-set to catch a row a peer changed meanwhile — so one session, but still a read after the yield. The memory store takes the same signature and re-reads its record after the await for the same reason. The router passes its asyncio.to_thread policy read as the callable and drops the separate probe; get stays on the protocol for tests and anchors. No short-TTL verdict cache: it would mask a successful login's reset for the TTL window, and the single session already halves the hot path. Sweep in the upsert's transaction, plus an index. record_failure ran the housekeeping DELETE in its own transaction and the upsert in a second one, two commits (two WAL flushes on PostgreSQL) per failed login; the sweep now rides in the upsert's transaction with one commit, the keep exclusion already protecting the row being recorded. The new ix_login_throttle_updated_at (ORM __table_args__ and the unreleased 0035 migration, created after the table and dropped before it, each guarded separately so a table left by an interrupted attempt still gets its index) serves the sweep's stale-counter predicate; the served-lock predicate scans a table the sweep itself keeps small. Tests (red first on a137fbc): a clean IP never resolves the policy and a recorded IP resolves it exactly once (both stores), the SQL store opens exactly one session per check and commits once per record_failure (counting wrappers), the migration test asserts the index after upgrade, after completing a partial upgrade that lacked it, and gone after downgrade while compare_metadata stays empty; the router-level "mutated during policy resolution" tests now inject their mutation through the policy callable. * docs(changelog): note config_version 57 in the login throttle entry
3 participants
Summary
Fixes #6471.
DbRunEventStorepreviously recomputed the nextseqfrom the maximum seq among surviving rows. Afterdelete_by_run(), that maximum could drop, so the next write reused an already-issued seq and became invisible to clients paginating withafter_seq.This change adds a durable per-thread high-water mark table,
run_event_thread_seq, and routes all DB write paths through it:put,put_batch, andput_if_absentallocate frommax(row_max, watermark) + 1.delete_by_run()preserves the current watermark before deleting rows.delete_by_thread(..., user_id=None)) removes the watermark so a recreated thread can restart at seq 1, matching the memory and JSONL backends.0034_run_event_seq_watermarkcreates and backfills the watermark from existingrun_events.Tests
118 passed, 4 skippedfor run-event store and migration-chain tests.109 passedfor the broader event-store, lock-lifecycle, pagination, filter, endpoint, and stream-contract tests.git diff --checkpass.