laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-12 09:13:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(quartz): reuse one subscription id across fetchAllPages pages

Browse Source 
Revert the fresh-subId-per-page workaround (ed5c25e2) now that the underlying
double-REQ race is fixed at the root in PoolRequests. Relays cap the number of
concurrent subscriptions per connection, so a single reused id — opened per page
with the page's `until`, closed before the next page — keeps the whole download
to one subscription slot instead of churning through a distinct id each page.

Safe because the pool now serializes the "send a REQ" decision: after a page's
EOSE, the auto-resend and the loop's unsubscribe+resubscribe can no longer both
fire a REQ for the same id (guarded by PoolRequestsConcurrencyTest). Unit suites
(negentropy paging scenarios, subscriptions) pass; full-scale real-relay
verification to follow before push.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z
This commit is contained in:
Claude
2026-07-01 13:01:12 +00:00
parent f357b445c3
commit 23722969fd
1 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt
+15 -8
View File
@@ -64,17 +64,24 @@ suspend fun INostrClient.fetchAllPages(
// Track how many matching events each filter has received so far.
val matchCountPerFilter = IntArray(filters.size)
// One subscription id reused for every page. Each page opens it (with the
// page's `until`), waits for EOSE, then closes it before the next page opens
// it again — so at most one subscription is ever live and the whole download
// occupies a single subscription slot on the connection (relays cap the
// number of concurrent subscriptions per connection, so churning through a
// fresh id per page is wasteful).
//
// Reusing the id is safe because the pool serializes the "send a REQ"
// decision: after each page's EOSE, the pool's auto-resend and this loop's
// unsubscribe+resubscribe can no longer both fire a REQ for the same id (see
// PoolRequests.decideCommandLocked / PoolRequestsConcurrencyTest). Without
// that fix the two raced and produced a duplicate REQ — two EOSEs, or an
// empty page that silently truncated large results.
val subId = newSubId()
while (true) {
coroutineContext.ensureActive()
// A fresh subscription id per page. Reusing one id across pages
// (unsubscribe + immediately re-subscribe the same id) races on the wire:
// in-flight events from the previous page's REQ bleed into the next page's
// listener. Those stale events carry a `created_at` above the new `until`,
// so `match()` rejects them, the page ends with `pageCount == 0`, and the
// whole download terminates early — silently truncating large results.
val subId = newSubId()
val pagedFilters =
if (until == null) {
filters