diff --git a/src/relay/pty-source-credit-ledger.test.ts b/src/relay/pty-source-credit-ledger.test.ts index 7f3759e9591..638676354aa 100644 --- a/src/relay/pty-source-credit-ledger.test.ts +++ b/src/relay/pty-source-credit-ledger.test.ts @@ -7,6 +7,57 @@ import { import { RelayPtySourceCreditLedger } from './pty-source-credit-ledger' import { CLOSED_DELIVERY_TOMBSTONE_LIMIT, type DeliveryRecord } from './pty-source-credit-record' +type BoundaryRecord = { sentBoundaries: Set } + +function getBoundaryRecord( + ledger: RelayPtySourceCreditLedger, + owner: PtySourceDeliveryIdentity +): BoundaryRecord { + const internals = ledger as unknown as { deliveries: Map } + const record = internals.deliveries.get(ptySourceDeliveryKey(owner)) + if (!record) { + throw new Error('test delivery record missing') + } + return record +} + +type CountingBoundarySet = Set & { readonly visits: number } + +function createCountingBoundarySet(boundaries: readonly number[]): CountingBoundarySet { + const pending = new Set(boundaries) + let visits = 0 + return { + get visits() { + return visits + }, + has: (boundary: number) => pending.has(boundary), + delete: (boundary: number) => pending.delete(boundary), + *[Symbol.iterator](): IterableIterator { + for (const boundary of pending) { + visits += 1 + yield boundary + } + } + } as unknown as CountingBoundarySet +} + +function countLegacyBoundaryVisits( + boundaries: readonly number[], + credits: readonly number[] +): number { + const pending = new Set(boundaries) + let visits = 0 + for (const creditedEndSu of credits) { + for (const boundary of pending) { + visits += 1 + if (boundary < creditedEndSu) { + pending.delete(boundary) + } + } + } + return visits +} + function identity( deliveryToken = 'token-1', overrides: Partial = {} @@ -239,6 +290,52 @@ describe('RelayPtySourceCreditLedger', () => { expect(ledger.reserveNextSend(owner, 2)!.span.data).toBe('ef') }) + it('stops ACK boundary cleanup at the first uncredited boundary', () => { + const boundaryCount = 1_024 + const spanCount = boundaryCount - 1 + const ledger = new RelayPtySourceCreditLedger({ + maxRetainedSpans: boundaryCount, + maxAggregateRetainedSpans: boundaryCount + }) + const owner = identity() + ledger.open(owner, boundaryCount) + for (let index = 0; index < spanCount; index += 1) { + append(ledger, owner, 'x', `span-${index}`) + const reservation = ledger.reserveNextSend(owner, 1) + expect(reservation).not.toBeNull() + ledger.commitSend(reservation!) + } + + const record = getBoundaryRecord(ledger, owner) + const insertionOrder = [...record.sentBoundaries] + expect(insertionOrder).toEqual(Array.from({ length: boundaryCount }, (_, index) => index)) + const countedBoundaries = createCountingBoundarySet(insertionOrder) + record.sentBoundaries = countedBoundaries + const credits = Array.from({ length: spanCount }, (_, index) => index + 1) + const legacyVisits = countLegacyBoundaryVisits(insertionOrder, credits) + + for (const creditedEndSu of credits) { + expect( + ledger.acknowledge(owner, { + id: owner.id, + clientGeneration: owner.clientGeneration, + ownerGeneration: owner.ownerGeneration, + deliveryToken: owner.deliveryToken, + creditedEndSu + }) + ).toBe('advanced') + } + + expect(legacyVisits).toBe(524_799) + expect(countedBoundaries.visits).toBe(2_046) + expect(legacyVisits - countedBoundaries.visits).toBe(522_753) + console.log( + `[bench] ${boundaryCount} sent boundaries: ACK iterator visits ${legacyVisits} -> ${countedBoundaries.visits} ` + + `(-${(((legacyVisits - countedBoundaries.visits) / legacyVisits) * 100).toFixed(2)}%)` + ) + expect(ledger.retentionSnapshot()).toEqual({ sourceSu: 0, dataBytes: 0, spans: 0 }) + }) + it('never exceeds a token source window across generated send/ACK sequences', () => { for (let seed = 1; seed <= 40; seed++) { const ledger = new RelayPtySourceCreditLedger() diff --git a/src/relay/pty-source-credit-settlement.ts b/src/relay/pty-source-credit-settlement.ts index e12e73521e5..55196110e79 100644 --- a/src/relay/pty-source-credit-settlement.ts +++ b/src/relay/pty-source-credit-settlement.ts @@ -31,9 +31,11 @@ function reclaimCreditedSpans(record: DeliveryRecord): void { function advanceCredit(record: DeliveryRecord, creditedEndSu: number): void { record.creditedEndSu = creditedEndSu for (const boundary of record.sentBoundaries) { - if (boundary < creditedEndSu) { - record.sentBoundaries.delete(boundary) + // Boundaries are inserted in increasing sentEndSu order, so later values cannot be eligible. + if (boundary >= creditedEndSu) { + break } + record.sentBoundaries.delete(boundary) } reclaimCreditedSpans(record) }