[mem] fix(memory): release consumed buffers, abandoned waits, and deleted diff views (#30650)

* fix(memory): release consumed stream buffers and abandoned request state

* fix(memory): preserve queued cancellation for shared promise callers

* fix(memory): keep workspace cache cleanup free of browser side effects

* fix(memory): follow renamed workspace owners in cached diff views

* fix(memory): a viewer unmounting after its workspace is removed cannot re-create cached diff entries

Combined-diff view state, scroll offset and scroll anchor now live in one owner-tagged
entry per view key. A mounted viewer claims its key; writes only update claimed entries,
so the late unmount write that follows a workspace removal is a no-op instead of
re-adding orphan scroll entries.

* fix(memory): tighten retention fixes and let evicted diff views heal

- Combined diff: writes go through a per-mount lease. Forgetting a workspace revokes its
  leases, so a late unmount write cannot resurrect entries, while an LRU-evicted entry of
  a still-mounted viewer is re-created on its next write. Drops the duplicated view scrollTop.
- NDJSON framer: copy a retained tail only when it is a strict slice of its chunk, copy an
  unfinished line once, and copy every rejected-line prefix.
- Binary frame reader: single exit path for the end-of-feed leftover copy.
- Relay readiness reuses waitForPromiseWithSignal instead of re-implementing it.
- PTY preconnect input: pass the listener-removing settle into createInput instead of
  mutating the input afterwards.

* docs(memory): state the deferSettled ordering guarantee precisely

---------

Co-authored-by: m4air <m4air@Mac.localdomain>
Co-authored-by: Neil <neil@stably.ai>
This commit is contained in:
OrcaWin
2026-10-19 16:23:32 -07:00
committed by GitHub
co-authored by m4air Neil
parent e7d6ee645b
commit ad5ef0d944
27 changed files with 829 additions and 148 deletions
@@ -162,6 +162,18 @@ export function createBinaryStreamFrameReader(
bufferedBytes -= bytes
}
const ownPendingHead = (): void => {
const head = chunks[0]
// Why: a leftover subarray pins its whole socket chunk; copying once per feed, not per frame, avoids quadratic batches.
if (
head &&
head.buffer.byteLength > Buffer.poolSize &&
head.length * 2 < head.buffer.byteLength
) {
chunks[0] = Buffer.from(head)
}
}
const deliver = (kind: number, body: Buffer): void => {
let event: unknown
try {
@@ -195,13 +207,14 @@ export function createBinaryStreamFrameReader(
}
const frameBytes = FRAME_HEADER_BYTES + bodyBytes
if (bufferedBytes < frameBytes) {
return
break
}
const head = contiguousHead(frameBytes)
const body = head.subarray(FRAME_HEADER_BYTES, frameBytes)
consume(head, frameBytes)
deliver(kind, body)
}
ownPendingHead()
}
}
}
@@ -0,0 +1,53 @@
import { describe, expect, it } from 'vitest'
import {
createBinaryStreamFrameReader,
encodeBinaryStreamEventFrame
} from './daemon-stream-binary-framing'
function collectBuffers(): number {
if (!globalThis.gc) {
throw new Error('Run this regression with --expose-gc')
}
globalThis.gc()
globalThis.gc()
return process.memoryUsage().arrayBuffers
}
describe('binary daemon stream retained buffers', () => {
it.each([2, 15])('releases consumed frames while keeping a %i-byte incomplete frame', (split) => {
const readers: { finish: () => void; count: () => number }[] = []
const baseline = collectBuffers()
for (let index = 0; index < 32; index += 1) {
let count = 0
const reader = createBinaryStreamFrameReader(
() => count++,
(error) => {
throw error
}
)
const next = encodeBinaryStreamEventFrame({ type: 'event', event: 'next' })
reader.feed(
Buffer.concat([
encodeBinaryStreamEventFrame({
type: 'event',
event: 'large',
payload: 'x'.repeat(1024 * 1024)
}),
next.subarray(0, split)
])
)
readers.push({
finish: () => reader.feed(next.subarray(split)),
count: () => count
})
}
const retainedBytes = collectBuffers() - baseline
expect(readers).toHaveLength(32)
expect(retainedBytes).toBeLessThan(8 * 1024 * 1024)
for (const reader of readers) {
expect(reader.count()).toBe(1)
reader.finish()
expect(reader.count()).toBe(2)
}
})
})
@@ -0,0 +1,54 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { sharedPromiseSettlementWaiters } from '../../../shared/promise-settlement-waiters'
import { RelayReadinessWaiters } from './relay-readiness'
afterEach(() => vi.useRealTimers())
describe('relay readiness wait lifetime', () => {
it('detaches expired mint callers while shared reconnect work remains pending', async () => {
vi.useFakeTimers()
let finish = (): void => {}
const work = new Promise<void>((resolve) => {
finish = resolve
})
const observed = vi.spyOn(work, 'then')
const waiters = new RelayReadinessWaiters()
const anchor = waiters.until(work, Date.now() + 10_000)
for (let index = 0; index < 100; index += 1) {
const waiting = waiters.until(work, Date.now() + 1)
await vi.advanceTimersByTimeAsync(1)
await expect(waiting).resolves.toBe(false)
}
expect(observed).toHaveBeenCalledOnce()
expect(sharedPromiseSettlementWaiters(work).waiterCount).toBe(1)
finish()
await expect(anchor).resolves.toBe(true)
expect(vi.getTimerCount()).toBe(0)
})
it('releases callers without cancelling shared work and permits a later wait', async () => {
vi.useFakeTimers()
let finish = (): void => {}
const work = new Promise<void>((resolve) => {
finish = resolve
})
const waiters = new RelayReadinessWaiters()
const first = waiters.until(work, Date.now() + 10_000)
waiters.releaseAll()
await expect(first).resolves.toBe(false)
expect(sharedPromiseSettlementWaiters(work).waiterCount).toBe(0)
const next = waiters.until(work, Date.now() + 10_000)
finish()
await expect(next).resolves.toBe(true)
expect(vi.getTimerCount()).toBe(0)
})
it('propagates reconnect failures and clears its deadline', async () => {
vi.useFakeTimers()
const error = new Error('reconnect failed')
await expect(
new RelayReadinessWaiters().until(Promise.reject(error), Date.now() + 10_000)
).rejects.toBe(error)
expect(vi.getTimerCount()).toBe(0)
})
})
@@ -1,4 +1,5 @@
import { RelayHttpError } from './relay-http-client'
import { waitForPromiseWithSignal } from '../../../shared/abort-signal-reason'
// Why the last reconcile could not produce a live broker. Every value is a
// lowercase token so `relay_<reason>` stays a valid mint failure code.
@@ -98,14 +99,19 @@ export class RelayReadinessWaiters {
// Resolves true when `work` settles first, false at the deadline or on release.
async until(work: Promise<unknown>, deadline: number): Promise<boolean> {
let release = (): void => {}
const released = new Promise<false>((resolve) => {
release = () => resolve(false)
})
const controller = new AbortController()
const released = new Error('Relay readiness wait released')
const release = (): void => controller.abort(released)
const timer = setTimeout(release, Math.max(0, deadline - Date.now()))
this.releases.add(release)
try {
return await Promise.race([work.then(() => true), released])
await waitForPromiseWithSignal(work, controller.signal)
return true
} catch (error) {
if (error === released) {
return false
}
throw error
} finally {
clearTimeout(timer)
this.releases.delete(release)
@@ -104,7 +104,8 @@ export default function CombinedDiffViewer({
setSectionHeights,
setSections,
setSideBySide: preferences.setSideBySide,
viewStateKey
viewStateKey,
worktreeId: file.worktreeId
})
const { loadSection, loadDeferredSection } = useCombinedDiffSectionLoader({
entrySet,
@@ -157,7 +158,7 @@ export default function CombinedDiffViewer({
sideBySide: preferences.sideBySide,
structureRevision: sectionRowKeys.structureRevision,
totalSize: virtualizer.getTotalSize(),
viewStateKey,
memoryLeaseRef: restore.memoryLeaseRef,
virtualizer
})
@@ -216,6 +217,7 @@ export default function CombinedDiffViewer({
entryCount: entrySet.entries.length,
entrySignature: entrySet.entrySignature,
loadedIndicesRef: registry.loadedIndicesRef,
memoryLeaseRef: restore.memoryLeaseRef,
scrollContainerRef,
sectionHeights,
sections,
@@ -227,6 +229,7 @@ export default function CombinedDiffViewer({
entrySignature: entrySet.entrySignature,
hasDirectScrollInput,
latestDomScrollAnchorRef: restore.latestDomScrollAnchorRef,
memoryLeaseRef: restore.memoryLeaseRef,
programmaticScrollMarks,
scrollAnchorRef: restore.scrollAnchorRef,
scrollContainerRef,
@@ -12,11 +12,7 @@ import { getLargeDiffRenderLimit } from '../../../../../../shared/large-diff-ren
import type { DiffSection } from '../../diff-section-types'
import type { CombinedDiffEntrySet } from '../resolve-changes/use-combined-diff-entry-set'
import { useCombinedDiffViewRestore } from '../remember-view/use-combined-diff-view-restore'
import {
combinedDiffScrollAnchorCache,
combinedDiffScrollTopCache,
combinedDiffViewStateCache
} from '../remember-view/combined-diff-view-memory'
import { clearCombinedDiffViewMemory } from '../remember-view/combined-diff-view-cache'
import { useCombinedDiffSectionLoadRegistry } from './combined-diff-section-load-registry'
import { useCombinedDiffSectionRetry } from './use-combined-diff-section-retry'
import { withDiffSectionLoadTimeout } from './combined-diff-section-load-timeout'
@@ -86,7 +82,8 @@ function useViewer(props: Viewer) {
setSectionHeights,
setSections,
setSideBySide,
viewStateKey: props.viewStateKey
viewStateKey: props.viewStateKey,
worktreeId: props.file.worktreeId
})
const loader = useCombinedDiffSectionLoader({
entrySet: props.entrySet,
@@ -201,9 +198,7 @@ afterEach(() => {
vi.restoreAllMocks()
vi.useRealTimers()
mocks.fetch.mockReset()
combinedDiffViewStateCache.clear()
combinedDiffScrollAnchorCache.clear()
combinedDiffScrollTopCache.clear()
clearCombinedDiffViewMemory()
})
describe('combined diff discarded-result analysis budget', () => {
@@ -0,0 +1,103 @@
import type { VirtualizedScrollAnchor } from '@/hooks/useVirtualizedScrollAnchor'
import { setWithLRU } from '@/lib/scroll-cache'
import type { DiffSection } from '../../diff-section-types'
export type CachedCombinedDiffViewState = {
entrySignature: string
gitStatusSignature: string
sections: DiffSection[]
sectionHeights: Record<number, number>
loadedIndices: number[]
sideBySide: boolean
}
type CombinedDiffViewMemory = {
worktreeId: string
view?: CachedCombinedDiffViewState
scrollTop?: number
scrollAnchor?: NonNullable<VirtualizedScrollAnchor>
}
/** A mounted viewer's write access to its key; revoked only when its workspace is forgotten. */
export type CombinedDiffViewLease = {
readonly viewKey: string
worktreeId: string
revoked: boolean
}
const memoryByViewKey = new Map<string, CombinedDiffViewMemory>()
const mountedLeases = new Set<CombinedDiffViewLease>()
export function readCombinedDiffViewMemory(
viewKey: string
): Readonly<CombinedDiffViewMemory> | undefined {
return memoryByViewKey.get(viewKey)
}
export function claimCombinedDiffView(viewKey: string, worktreeId: string): CombinedDiffViewLease {
if (memoryByViewKey.get(viewKey)?.worktreeId !== worktreeId) {
memoryByViewKey.delete(viewKey)
}
const lease = { viewKey, worktreeId, revoked: false }
mountedLeases.add(lease)
return lease
}
// Why: release only stops tracking — the viewer's unmount snapshot still writes through the lease afterwards.
export function releaseCombinedDiffView(lease: CombinedDiffViewLease): void {
mountedLeases.delete(lease)
}
export function rememberCombinedDiffView(
lease: CombinedDiffViewLease | null,
patch: Partial<Omit<CombinedDiffViewMemory, 'worktreeId'>>
): void {
if (!lease || lease.revoked) {
return
}
const memory = memoryByViewKey.get(lease.viewKey)
const owned = memory?.worktreeId === lease.worktreeId ? memory : { worktreeId: lease.worktreeId }
setWithLRU(memoryByViewKey, lease.viewKey, { ...owned, ...patch })
}
export function forgetCombinedDiffViewsShowingPath(relativePath: string): void {
for (const [viewKey, memory] of memoryByViewKey) {
if (memory.view?.sections.some((section) => section.path === relativePath)) {
memoryByViewKey.set(viewKey, { ...memory, view: undefined })
}
}
}
export function renameCombinedDiffViewOwner(oldWorktreeId: string, newWorktreeId: string): void {
for (const [viewKey, memory] of memoryByViewKey) {
if (memory.worktreeId === oldWorktreeId) {
memoryByViewKey.set(viewKey, { ...memory, worktreeId: newWorktreeId })
}
}
for (const lease of mountedLeases) {
if (lease.worktreeId === oldWorktreeId) {
lease.worktreeId = newWorktreeId
}
}
}
/** A deleted workspace cannot restore its cached file bodies. Ordinary tab closes keep them. */
export function forgetCombinedDiffViewsForWorktrees(worktreeIds: Iterable<string>): void {
const removed = new Set(worktreeIds)
for (const [viewKey, memory] of memoryByViewKey) {
if (removed.has(memory.worktreeId)) {
memoryByViewKey.delete(viewKey)
}
}
for (const lease of mountedLeases) {
if (removed.has(lease.worktreeId)) {
lease.revoked = true
mountedLeases.delete(lease)
}
}
}
export function clearCombinedDiffViewMemory(): void {
memoryByViewKey.clear()
mountedLeases.clear()
}
@@ -1,23 +1,8 @@
import type { VirtualizedScrollAnchor } from '@/hooks/useVirtualizedScrollAnchor'
import type { DiffSection } from '../../diff-section-types'
import {
ORCA_EDITOR_EXTERNAL_FILE_CHANGE_EVENT,
type EditorPathMutationTarget
} from '../../editor-autosave'
export type CachedCombinedDiffViewState = {
entrySignature: string
gitStatusSignature: string
sections: DiffSection[]
sectionHeights: Record<number, number>
loadedIndices: number[]
scrollTop: number
sideBySide: boolean
}
export const combinedDiffViewStateCache = new Map<string, CachedCombinedDiffViewState>()
export const combinedDiffScrollTopCache = new Map<string, number>()
export const combinedDiffScrollAnchorCache = new Map<string, VirtualizedScrollAnchor>()
import { forgetCombinedDiffViewsShowingPath } from './combined-diff-view-cache'
// Why: session-scoped toolbar choices outlive the unmount, so they are module state rather than component state.
export const combinedDiffViewPreferences: {
@@ -26,19 +11,11 @@ export const combinedDiffViewPreferences: {
fileTreeCollapsed: boolean | null
} = { collapsed: null, sideBySide: null, fileTreeCollapsed: null }
function invalidateCombinedDiffCachesForRelativePath(relativePath: string): void {
for (const [key, cached] of combinedDiffViewStateCache.entries()) {
if (cached.sections.some((section) => section.path === relativePath)) {
combinedDiffViewStateCache.delete(key)
}
}
}
function handleCombinedDiffExternalFileChange(event: Event): void {
const detail = (event as CustomEvent<EditorPathMutationTarget>).detail
if (detail?.relativePath) {
// Why: inactive combined-diff tabs are unmounted, so only a module-level cache bust stops a remount replaying stale bodies.
invalidateCombinedDiffCachesForRelativePath(detail.relativePath)
forgetCombinedDiffViewsShowingPath(detail.relativePath)
}
}
@@ -1,13 +1,17 @@
import { useEffect } from 'react'
import type React from 'react'
import { setWithLRU } from '@/lib/scroll-cache'
import type { DiffSection } from '../../diff-section-types'
import { combinedDiffScrollTopCache, combinedDiffViewStateCache } from './combined-diff-view-memory'
import {
readCombinedDiffViewMemory,
rememberCombinedDiffView,
type CombinedDiffViewLease
} from './combined-diff-view-cache'
export function useCombinedDiffViewPersist({
combinedGitStatusSignature,
entryCount,
entrySignature,
memoryLeaseRef,
scrollContainerRef,
sectionHeights,
sections,
@@ -18,6 +22,7 @@ export function useCombinedDiffViewPersist({
combinedGitStatusSignature: string
entryCount: number
entrySignature: string
memoryLeaseRef: React.RefObject<CombinedDiffViewLease | null>
scrollContainerRef: React.RefObject<HTMLDivElement | null>
sectionHeights: Record<number, number>
sections: DiffSection[]
@@ -29,24 +34,28 @@ export function useCombinedDiffViewPersist({
if (sections.length === 0 && entryCount > 0) {
return
}
const preservedScrollTop =
combinedDiffScrollTopCache.get(viewStateKey) ?? scrollContainerRef.current?.scrollTop ?? 0
setWithLRU(combinedDiffViewStateCache, viewStateKey, {
entrySignature,
gitStatusSignature: combinedGitStatusSignature,
sections,
sectionHeights,
loadedIndices: Array.from(loadedIndicesRef.current).filter(
(index) => !sections[index]?.loading
),
scrollTop: preservedScrollTop,
sideBySide
rememberCombinedDiffView(memoryLeaseRef.current, {
scrollTop:
readCombinedDiffViewMemory(viewStateKey)?.scrollTop ??
scrollContainerRef.current?.scrollTop ??
0,
view: {
entrySignature,
gitStatusSignature: combinedGitStatusSignature,
sections,
sectionHeights,
loadedIndices: Array.from(loadedIndicesRef.current).filter(
(index) => !sections[index]?.loading
),
sideBySide
}
})
}, [
combinedGitStatusSignature,
entryCount,
entrySignature,
loadedIndicesRef,
memoryLeaseRef,
scrollContainerRef,
sectionHeights,
sections,
@@ -7,7 +7,12 @@ import type { GitStatusEntry } from '../../../../../../shared/git-status-types'
import type { DiffSection } from '../../diff-section-types'
import { useCombinedDiffSectionLoadRegistry } from '../load-sections/combined-diff-section-load-registry'
import type { CombinedDiffEntrySet } from '../resolve-changes/use-combined-diff-entry-set'
import { combinedDiffViewStateCache } from './combined-diff-view-memory'
import {
claimCombinedDiffView,
clearCombinedDiffViewMemory,
releaseCombinedDiffView,
rememberCombinedDiffView
} from './combined-diff-view-cache'
import { useCombinedDiffViewRestore } from './use-combined-diff-view-restore'
import { createStore } from 'zustand/vanilla'
import { createEditorModelRegistry } from '@/lib/editor-model-registry'
@@ -51,7 +56,8 @@ function restoreSections(entrySet: CombinedDiffEntrySet, viewStateKey: string):
sections = typeof value === 'function' ? value(sections) : value
},
setSideBySide: () => {},
viewStateKey
viewStateKey,
worktreeId: 'wt-1'
})
})
return sections
@@ -60,7 +66,7 @@ function restoreSections(entrySet: CombinedDiffEntrySet, viewStateKey: string):
describe('useCombinedDiffViewRestore deferral', () => {
afterEach(() => {
cleanup()
combinedDiffViewStateCache.clear()
clearCombinedDiffViewMemory()
})
it('keeps uncommitted rows deferred when only the branch pass counted', () => {
@@ -105,15 +111,18 @@ describe('useCombinedDiffViewRestore deferral', () => {
largeDiffRenderLimit: null
}
]
combinedDiffViewStateCache.set(viewStateKey, {
entrySignature: entrySet.entrySignature,
gitStatusSignature: '',
sections: loadedSections,
sectionHeights: {},
loadedIndices: [0],
scrollTop: 0,
sideBySide: false
const lease = claimCombinedDiffView(viewStateKey, 'wt-1')
rememberCombinedDiffView(lease, {
view: {
entrySignature: entrySet.entrySignature,
gitStatusSignature: '',
sections: loadedSections,
sectionHeights: {},
loadedIndices: [0],
sideBySide: false
}
})
releaseCombinedDiffView(lease)
cleanup()
const closedTab: OpenFile = {
@@ -15,16 +15,19 @@ import {
import type { CombinedDiffEntrySet } from '../resolve-changes/use-combined-diff-entry-set'
import type { CombinedDiffSectionLoadRegistry } from '../load-sections/combined-diff-section-load-registry'
import { clearPendingSectionReloadTimers } from '../load-sections/combined-diff-section-load-registry'
import { combinedDiffViewPreferences } from './combined-diff-view-memory'
import {
combinedDiffScrollAnchorCache,
combinedDiffScrollTopCache,
combinedDiffViewPreferences,
combinedDiffViewStateCache
} from './combined-diff-view-memory'
claimCombinedDiffView,
readCombinedDiffViewMemory,
releaseCombinedDiffView,
rememberCombinedDiffView,
type CombinedDiffViewLease
} from './combined-diff-view-cache'
export type CombinedDiffViewRestore = {
invalidateViewStateCache: () => void
latestDomScrollAnchorRef: React.RefObject<VirtualizedScrollAnchor>
memoryLeaseRef: React.RefObject<CombinedDiffViewLease | null>
scrollAnchorRef: React.RefObject<VirtualizedScrollAnchor>
scrollOffsetRef: React.RefObject<number>
}
@@ -37,7 +40,8 @@ export function useCombinedDiffViewRestore({
setSectionHeights,
setSections,
setSideBySide,
viewStateKey
viewStateKey,
worktreeId
}: {
entrySet: CombinedDiffEntrySet
gitStatusEntries: GitStatusEntry[]
@@ -47,6 +51,7 @@ export function useCombinedDiffViewRestore({
setSections: React.Dispatch<React.SetStateAction<DiffSection[]>>
setSideBySide: React.Dispatch<React.SetStateAction<boolean>>
viewStateKey: string
worktreeId: string
}): CombinedDiffViewRestore {
const {
entries,
@@ -66,17 +71,25 @@ export function useCombinedDiffViewRestore({
supersededLoadIndicesRef
} = registry
// Why useState and not `useRef(expr)`: the latter re-reads all three caches on every render and
// Why useState and not `useRef(expr)`: the latter re-reads the cache on every render and
// throws the result away, and an anchor seeds legitimately to null so a nullish guard would keep
// re-reading. useState's initializer runs once without writing a ref during render.
const [restoreSeed] = useState<{ offset: number; anchor: VirtualizedScrollAnchor }>(() => ({
offset: combinedDiffScrollTopCache.get(viewStateKey) ?? 0,
anchor: combinedDiffScrollAnchorCache.get(viewStateKey) ?? null
}))
const [restoreSeed] = useState<{ offset: number; anchor: VirtualizedScrollAnchor }>(() => {
const memory = readCombinedDiffViewMemory(viewStateKey)
return { offset: memory?.scrollTop ?? 0, anchor: memory?.scrollAnchor ?? null }
})
const scrollOffsetRef = useRef(restoreSeed.offset)
const scrollAnchorRef = useRef<VirtualizedScrollAnchor>(restoreSeed.anchor)
const latestDomScrollAnchorRef = useRef<VirtualizedScrollAnchor>(restoreSeed.anchor)
// Why: keep above CombinedDiffViewer's persist/scroll hooks so the lease exists before their effects write.
const memoryLeaseRef = useRef<CombinedDiffViewLease | null>(null)
useLayoutEffect(() => {
const lease = claimCombinedDiffView(viewStateKey, worktreeId)
memoryLeaseRef.current = lease
return () => releaseCombinedDiffView(lease)
}, [viewStateKey, worktreeId])
// Why: tab/worktree switches unmount this viewer; cache by pane key so remount restores sections+scroll before repaint.
const initializedEntryStateRef = useRef<{
viewStateKey: string
@@ -98,7 +111,8 @@ export function useCombinedDiffViewRestore({
hasUncommittedEntriesSnapshot
}
deferredLoadRequestsRef.current.clear()
const cached = combinedDiffViewStateCache.get(viewStateKey)
const memory = readCombinedDiffViewMemory(viewStateKey)
const cached = memory?.view
const canRestoreSnapshotSectionsByKey =
hasUncommittedEntriesSnapshot &&
cached !== undefined &&
@@ -133,14 +147,14 @@ export function useCombinedDiffViewRestore({
})
)
loadingIndicesRef.current.clear()
scrollOffsetRef.current = combinedDiffScrollTopCache.get(viewStateKey) ?? cached.scrollTop
scrollAnchorRef.current = combinedDiffScrollAnchorCache.get(viewStateKey) ?? null
scrollOffsetRef.current = memory.scrollTop ?? 0
scrollAnchorRef.current = memory.scrollAnchor ?? null
latestDomScrollAnchorRef.current = scrollAnchorRef.current
return
}
scrollOffsetRef.current = combinedDiffScrollTopCache.get(viewStateKey) ?? 0
scrollAnchorRef.current = combinedDiffScrollAnchorCache.get(viewStateKey) ?? null
scrollOffsetRef.current = memory?.scrollTop ?? 0
scrollAnchorRef.current = memory?.scrollAnchor ?? null
latestDomScrollAnchorRef.current = scrollAnchorRef.current
// Why: separates "this row is uncounted" from "this pass skipped counting",
// which decides whether an uncounted row is cheap. Per pass, not per view:
@@ -209,12 +223,13 @@ export function useCombinedDiffViewRestore({
])
const invalidateViewStateCache = useCallback((): void => {
combinedDiffViewStateCache.delete(viewStateKey)
}, [viewStateKey])
rememberCombinedDiffView(memoryLeaseRef.current, { view: undefined })
}, [])
return {
invalidateViewStateCache,
latestDomScrollAnchorRef,
memoryLeaseRef,
scrollAnchorRef,
scrollOffsetRef
}
@@ -95,7 +95,7 @@ function useCombinedDiffSectionPasses({
sideBySide: false,
structureRevision: sectionRowKeys.structureRevision,
totalSize: 0,
viewStateKey: 'scaling-test',
memoryLeaseRef: { current: null },
virtualizer: FAKE_VIRTUALIZER
})
const treeNavigation = useCombinedDiffTreeNavigation({
@@ -99,7 +99,7 @@ function renderSignals(steps: readonly Step[]): (string | undefined)[] {
sideBySide: step.sideBySide,
structureRevision: rowKeys.structureRevision,
totalSize: 0,
viewStateKey: 'equivalence-test',
memoryLeaseRef: { current: null },
virtualizer: {
getVirtualItems: () => [],
isScrolling: false,
@@ -8,9 +8,11 @@ import {
} from '@/hooks/useVirtualizedScrollAnchor'
import { getVirtualizedScrollAnchorForOffset } from '@/hooks/virtualized-scroll-anchor-recording'
import type { ProgrammaticScrollMarks } from '@/hooks/programmatic-scroll-marks'
import { setWithLRU } from '@/lib/scroll-cache'
import type { DiffSection } from '../../diff-section-types'
import { combinedDiffScrollAnchorCache } from '../remember-view/combined-diff-view-memory'
import {
rememberCombinedDiffView,
type CombinedDiffViewLease
} from '../remember-view/combined-diff-view-cache'
export type CombinedDiffScrollAnchors = {
persistScrollAnchor: (refreshDomAnchor?: boolean) => void
@@ -35,7 +37,7 @@ export function useCombinedDiffScrollAnchors({
sideBySide,
structureRevision,
totalSize,
viewStateKey,
memoryLeaseRef,
virtualizer
}: {
clampRestoreCount: number
@@ -52,7 +54,7 @@ export function useCombinedDiffScrollAnchors({
sideBySide: boolean
structureRevision: number
totalSize: number
viewStateKey: string
memoryLeaseRef: React.RefObject<CombinedDiffViewLease | null>
virtualizer: Virtualizer<HTMLDivElement, Element>
}): CombinedDiffScrollAnchors {
const getCombinedDiffSectionKey = useCallback((section: DiffSection): string => section.key, [])
@@ -120,13 +122,10 @@ export function useCombinedDiffScrollAnchors({
return true
}, [latestDomScrollAnchorRef, scrollAnchorRef, scrollContainerRef])
const writeScrollAnchor = useCallback((): void => {
const anchor = scrollAnchorRef.current
if (anchor) {
setWithLRU(combinedDiffScrollAnchorCache, viewStateKey, anchor)
} else {
combinedDiffScrollAnchorCache.delete(viewStateKey)
}
}, [scrollAnchorRef, viewStateKey])
rememberCombinedDiffView(memoryLeaseRef.current, {
scrollAnchor: scrollAnchorRef.current ?? undefined
})
}, [memoryLeaseRef, scrollAnchorRef])
const persistScrollAnchor = useCallback(
(refreshDomAnchor = true): void => {
if (refreshDomAnchor) {
@@ -2,12 +2,12 @@ import { useLayoutEffect } from 'react'
import type React from 'react'
import type { VirtualizedScrollAnchor } from '@/hooks/useVirtualizedScrollAnchor'
import type { ProgrammaticScrollMarks } from '@/hooks/programmatic-scroll-marks'
import { setWithLRU } from '@/lib/scroll-cache'
import type { DiffSection } from '../../diff-section-types'
import {
combinedDiffScrollTopCache,
combinedDiffViewStateCache
} from '../remember-view/combined-diff-view-memory'
readCombinedDiffViewMemory,
rememberCombinedDiffView,
type CombinedDiffViewLease
} from '../remember-view/combined-diff-view-cache'
import type { CombinedDiffScrollAnchors } from './use-combined-diff-scroll-anchors'
export function useCombinedDiffScrollPersistence({
@@ -15,6 +15,7 @@ export function useCombinedDiffScrollPersistence({
entrySignature,
hasDirectScrollInput,
latestDomScrollAnchorRef,
memoryLeaseRef,
programmaticScrollMarks,
scrollAnchorRef,
scrollContainerRef,
@@ -30,6 +31,7 @@ export function useCombinedDiffScrollPersistence({
entrySignature: string
hasDirectScrollInput: () => boolean
latestDomScrollAnchorRef: React.RefObject<VirtualizedScrollAnchor>
memoryLeaseRef: React.RefObject<CombinedDiffViewLease | null>
programmaticScrollMarks: ProgrammaticScrollMarks
scrollAnchorRef: React.RefObject<VirtualizedScrollAnchor>
scrollContainerRef: React.RefObject<HTMLDivElement | null>
@@ -49,9 +51,9 @@ export function useCombinedDiffScrollPersistence({
return
}
const cached = combinedDiffViewStateCache.get(viewStateKey)
if (cached && cached.entrySignature === entrySignature) {
scrollOffsetRef.current = combinedDiffScrollTopCache.get(viewStateKey) ?? cached.scrollTop
const memory = readCombinedDiffViewMemory(viewStateKey)
if (memory?.view && memory.view.entrySignature === entrySignature) {
scrollOffsetRef.current = memory.scrollTop ?? 0
}
let anchorIdleTimerId: number | null = null
@@ -93,9 +95,7 @@ export function useCombinedDiffScrollPersistence({
scrollTop: number
writeAnchor: boolean
}): void => {
const existing = combinedDiffViewStateCache.get(viewStateKey)
scrollOffsetRef.current = scrollTop
setWithLRU(combinedDiffScrollTopCache, viewStateKey, scrollTop)
if (writeAnchor) {
if (recordDomAnchor) {
persistScrollAnchor()
@@ -107,13 +107,7 @@ export function useCombinedDiffScrollPersistence({
scheduleSettledAnchorPersist()
}
updateScrollbar()
if (!existing || existing.entrySignature !== entrySignature) {
return
}
setWithLRU(combinedDiffViewStateCache, viewStateKey, {
...existing,
scrollTop
})
rememberCombinedDiffView(memoryLeaseRef.current, { scrollTop })
}
let lastScrollHeight = container.scrollHeight
const handleScroll = (event: Event): void => {
@@ -165,6 +159,7 @@ export function useCombinedDiffScrollPersistence({
entrySignature,
hasDirectScrollInput,
latestDomScrollAnchorRef,
memoryLeaseRef,
persistScrollAnchor,
programmaticScrollMarks,
recordVirtualScrollAnchor,
@@ -273,12 +273,18 @@ export function createPtyPreconnectInputBuffer(
resolve(false)
return
}
const input = createInput(data, 'accepted', inputKind, resolve, signal)
const onAbort = (): void => withdraw(input)
// Why: a long-lived caller signal would otherwise pin every settled input through its listener.
const settle = (accepted: boolean): void => {
signal?.removeEventListener('abort', onAbort)
resolve(accepted)
}
const input = createInput(data, 'accepted', inputKind, settle, signal)
if (!retain(input)) {
resolve(false)
return
}
signal?.addEventListener('abort', () => withdraw(input), { once: true })
signal?.addEventListener('abort', onAbort, { once: true })
notifyRetained(onRetained, input)
})
},
@@ -0,0 +1,80 @@
import { getEventListeners } from 'node:events'
import { describe, expect, it } from 'vitest'
import { createPtyPreconnectInputBuffer } from './pty-preconnect-input-buffer'
import { createDeferred } from './pty-connection-test-async'
describe('preconnect input lifetime', () => {
it('releases abort listeners after repeated successful flushes', async () => {
const controller = new AbortController()
const delivered: string[] = []
for (let index = 0; index < 50; index += 1) {
const buffer = createPtyPreconnectInputBuffer()
const text = `paste-${index}`
const accepted = buffer.enqueueAccepted(text, 'driving', undefined, {
signal: controller.signal
})
await buffer.flush({
isCurrent: () => true,
sendInput: () => true,
sendInputImmediate: () => true,
sendInputAccepted: async (data) => {
delivered.push(data)
return true
}
})
await expect(accepted).resolves.toBe(true)
expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
}
expect(delivered).toHaveLength(50)
controller.abort()
expect(delivered).toHaveLength(50)
})
it('releases queued and active paste listeners on clear while a write is stalled', async () => {
const controller = new AbortController()
const buffer = createPtyPreconnectInputBuffer()
const started = createDeferred<void>()
const writing = createDeferred<boolean>()
const first = buffer.enqueueAccepted('active', 'driving', undefined, {
signal: controller.signal
})
const second = buffer.enqueueAccepted('queued', 'driving', undefined, {
signal: controller.signal
})
const flush = buffer.flush({
isCurrent: () => true,
sendInput: () => true,
sendInputImmediate: () => true,
sendInputAccepted: () => {
started.resolve()
return writing.promise
}
})
await started.promise
buffer.clear()
await expect(first).resolves.toBe(false)
await expect(second).resolves.toBe(false)
await flush
expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
writing.resolve(true)
})
it('withdraws an aborted paste without dropping the following input', async () => {
const controller = new AbortController()
const buffer = createPtyPreconnectInputBuffer()
const accepted = buffer.enqueueAccepted('cancelled', 'driving', undefined, {
signal: controller.signal
})
buffer.enqueue('following', 'ordinary', 'driving')
controller.abort()
await expect(accepted).resolves.toBe(false)
const delivered: string[] = []
await buffer.flush({
isCurrent: () => true,
sendInput: (text) => (delivered.push(text), true),
sendInputImmediate: () => true
})
expect(delivered).toEqual(['following'])
expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
})
})
@@ -0,0 +1,146 @@
import { afterEach, describe, expect, it } from 'vitest'
import { hostWorktreeTarget } from '../../../../shared/worktree/host-qualified-identity'
import { folderWorkspaceKey } from '../../../../shared/workspace-scope'
import {
claimCombinedDiffView,
clearCombinedDiffViewMemory,
readCombinedDiffViewMemory,
releaseCombinedDiffView,
rememberCombinedDiffView,
type CombinedDiffViewLease
} from '@/components/editor/combined-diff/remember-view/combined-diff-view-cache'
import { createTestStore, makeWorktree } from './store-test-helpers'
import { buildWorktreePurgeState } from './worktrees/teardown/worktree-purge-state'
import { applyRemoveWorktreeSuccessState } from './worktrees/teardown/remove-worktree-store-cleanup'
import { buildWorktreeRenameState } from './worktrees/session/worktree-identity-rename-state'
const removed = 'repo::/workspace/deleted'
const kept = 'repo::/workspace/kept'
function mountDiff(worktreeId: string): { key: string; lease: CombinedDiffViewLease } {
const key = `${worktreeId}::all-diffs::uncommitted::pane`
const lease = claimCombinedDiffView(key, worktreeId)
rememberCombinedDiffView(lease, {
view: {
entrySignature: 'entries',
gitStatusSignature: '',
sections: [
{
key: 'source',
path: 'source.ts',
status: 'modified',
originalContent: 'before'.repeat(1000),
modifiedContent: 'after'.repeat(1000),
collapsed: false,
loading: false,
dirty: false,
diffResult: null,
largeDiffRenderLimit: null
}
],
sectionHeights: {},
loadedIndices: [0],
sideBySide: true
},
scrollTop: 42,
scrollAnchor: { key: 'source', offset: 0 }
})
return { key, lease }
}
function seedDiff(worktreeId: string): string {
const { key, lease } = mountDiff(worktreeId)
releaseCombinedDiffView(lease)
return key
}
afterEach(() => {
clearCombinedDiffViewMemory()
})
describe('removed workspace diff memory', () => {
it('preserves cached bodies and scroll through rename, then releases them on deletion', () => {
const key = seedDiff(removed)
const keptKey = seedDiff(kept)
const renamed = 'repo::/workspace/renamed'
const store = createTestStore()
const sections = readCombinedDiffViewMemory(key)?.view?.sections
store.setState(buildWorktreeRenameState(store.getState(), removed, renamed))
expect(readCombinedDiffViewMemory(key)).toMatchObject({
worktreeId: renamed,
scrollTop: 42,
scrollAnchor: { key: 'source', offset: 0 }
})
expect(readCombinedDiffViewMemory(key)?.view?.sections).toBe(sections)
buildWorktreePurgeState(store.getState(), [removed])
expect(readCombinedDiffViewMemory(key)).toBeDefined()
buildWorktreePurgeState(store.getState(), [renamed])
expect(readCombinedDiffViewMemory(key)).toBeUndefined()
expect(readCombinedDiffViewMemory(keptKey)).toBeDefined()
})
it.each(['listing', 'explicit'] as const)(
'releases cached file bodies on %s removal and preserves another workspace',
(path) => {
const removedKey = seedDiff(removed)
const keptKey = seedDiff(kept)
const overlappingKey = seedDiff(`${removed}::all-diffs::nested-directory`)
const store = createTestStore()
if (path === 'listing') {
store.setState(buildWorktreePurgeState(store.getState(), [removed]))
} else {
applyRemoveWorktreeSuccessState(store.setState, removed, new Set())
}
expect(readCombinedDiffViewMemory(removedKey)).toBeUndefined()
expect(readCombinedDiffViewMemory(keptKey)).toBeDefined()
expect(readCombinedDiffViewMemory(overlappingKey)).toBeDefined()
}
)
it('ignores the unmount write of a viewer whose workspace was already removed', () => {
const { key, lease } = mountDiff(removed)
const store = createTestStore()
applyRemoveWorktreeSuccessState(store.setState, removed, new Set())
releaseCombinedDiffView(lease)
rememberCombinedDiffView(lease, { scrollTop: 7, scrollAnchor: { key: 'source', offset: 3 } })
expect(readCombinedDiffViewMemory(key)).toBeUndefined()
// A workspace recreated under the same id starts fresh once its viewer mounts.
rememberCombinedDiffView(claimCombinedDiffView(key, removed), { scrollTop: 7 })
expect(readCombinedDiffViewMemory(key)).toEqual({ worktreeId: removed, scrollTop: 7 })
})
it('lets a mounted viewer re-create its entry after LRU eviction', () => {
const { key, lease } = mountDiff(kept)
for (let index = 0; index < 20; index += 1) {
seedDiff(`repo::/workspace/other-${index}`)
}
expect(readCombinedDiffViewMemory(key)).toBeUndefined()
rememberCombinedDiffView(lease, { scrollTop: 9 })
expect(readCombinedDiffViewMemory(key)).toEqual({ worktreeId: kept, scrollTop: 9 })
})
it('keeps cached views when a different host still owns the same workspace id', () => {
const key = seedDiff(removed)
const store = createTestStore()
store.setState({
worktreesByRepo: {
repo: [
makeWorktree({ id: removed, repoId: 'repo', hostId: 'local' }),
makeWorktree({ id: removed, repoId: 'repo', hostId: 'runtime:remote' })
]
}
})
buildWorktreePurgeState(store.getState(), [
hostWorktreeTarget({ id: removed, hostId: 'runtime:remote' })
])
expect(readCombinedDiffViewMemory(key)?.scrollTop).toBe(42)
})
it('releases folder workspace views using their recorded owner', () => {
const workspaceId = folderWorkspaceKey('folder-1')
const key = seedDiff(workspaceId)
const store = createTestStore()
buildWorktreePurgeState(store.getState(), [workspaceId])
expect(readCombinedDiffViewMemory(key)).toBeUndefined()
})
})
@@ -12,6 +12,7 @@ import {
type ClosedTerminalTabSnapshot
} from '../../recently-closed-tabs'
import { renameWorktreeLayoutKeys } from '../../tabs/worktree-layout-lifecycle'
import { renameCombinedDiffViewOwner } from '@/components/editor/combined-diff/remember-view/combined-diff-view-cache'
// Every worktree-id-keyed store map the rename path re-keys, so a new `*ByWorktree` map isn't silently missed.
// Layout maps are re-keyed by renameWorktreeLayoutKeys. Tab-id/file-id-keyed maps are deliberately
@@ -57,6 +58,7 @@ export function buildWorktreeRenameState(
if (oldWorktreeId === newWorktreeId) {
return {}
}
renameCombinedDiffViewOwner(oldWorktreeId, newWorktreeId)
const renamed: Record<string, unknown> = {}
const renameKey = <T>(
key: keyof AppState,
@@ -1,4 +1,5 @@
import { worktreeWorkspaceKey } from '../../../../../../shared/workspace-scope'
import { forgetCombinedDiffViewsForWorktrees } from '@/components/editor/combined-diff/remember-view/combined-diff-view-cache'
import type { ExecutionHostId } from '../../../../../../shared/execution-host'
import type { WorktreeSliceSet } from '../listing/worktree-slice-types'
import { removeDeleteStatesForWorktreeIds } from './worktree-delete-state'
@@ -18,6 +19,7 @@ export function applyRemoveWorktreeSuccessState(
// Why outside `set`: it is module-scope, not store state. Dropping it also
// re-arms the once-per-workspace warning if this id is ever added back.
forgetAmbiguousOwnerWarnings([worktreeId])
forgetCombinedDiffViewsForWorktrees([worktreeId])
set((s) => {
const worktreeIds = [worktreeId]
const omitByWorktree = <T>(m: Record<string, T> | undefined) => omitRecordKeys(m, worktreeIds)
@@ -1,4 +1,5 @@
import type { AppState } from '../../../types'
import { forgetCombinedDiffViewsForWorktrees } from '@/components/editor/combined-diff/remember-view/combined-diff-view-cache'
import type { WorktreePurgeTarget, WorktreePurgeTargets } from '../../worktree-helpers'
import { forgetHugeRepoWarningDismissalsForWorktrees } from '@/lib/source-control-huge-repo-warning-dismissals'
import { parseWorkspaceKey } from '../../../../../../shared/workspace-scope'
@@ -69,6 +70,7 @@ export function buildFullWorktreePurgeState(
normalizedTargets: readonly WorktreePurgeTarget[]
): Partial<AppState> {
const worktreeIdSet = new Set(normalizedTargets.map((target) => target.id))
forgetCombinedDiffViewsForWorktrees(worktreeIdSet)
// Why: bulk purges bypass close actions, so provisional ownership must die before tab state does.
const cancelledSessionIds = cancelStructuredLaunchesForWorktrees(worktreeIdSet)
for (const worktreeId of worktreeIdSet) {
@@ -0,0 +1,34 @@
import { expect, it } from 'vitest'
import { waitForPromiseWithSignal } from './abort-signal-reason'
async function cancelCallers(work: Promise<void>): Promise<WeakRef<Uint8Array>[]> {
const references: WeakRef<Uint8Array>[] = []
for (let index = 0; index < 32; index += 1) {
const payload = new Uint8Array(1024 * 1024)
references.push(new WeakRef(payload))
const controller = new AbortController()
const waiting = waitForPromiseWithSignal(work, controller.signal).catch(() => undefined)
controller.abort(new Error('request cancelled', { cause: payload }))
await waiting
}
return references
}
it('collects cancelled callers and their payloads before shared work finishes', async () => {
let finish = (): void => {}
const work = new Promise<void>((resolve) => {
finish = resolve
})
const anchor = waitForPromiseWithSignal(work, new AbortController().signal)
const references = await cancelCallers(work)
await new Promise<void>((resolve) => setImmediate(resolve))
if (!globalThis.gc) {
throw new Error('Run this regression with --expose-gc')
}
globalThis.gc()
globalThis.gc()
const retained = references.filter((reference) => reference.deref() !== undefined).length
finish()
await anchor
expect(retained).toBe(0)
})
+62 -1
View File
@@ -1,5 +1,7 @@
import { expect, it } from 'vitest'
import { getEventListeners } from 'node:events'
import { expect, it, vi } from 'vitest'
import { waitForPromiseWithSignal } from './abort-signal-reason'
import { sharedPromiseSettlementWaiters } from './promise-settlement-waiters'
it('observes work that rejects after its caller has already canceled', async () => {
const controller = new AbortController()
@@ -12,3 +14,62 @@ it('observes work that rejects after its caller has already canceled', async ()
await expect(waitForPromiseWithSignal(work, controller.signal)).rejects.toBe(reason)
await new Promise<void>((resolve) => setImmediate(resolve))
})
it('detaches cancelled callers from shared stalled work while another caller waits', async () => {
let finish = (_value: number): void => {}
const work = new Promise<number>((resolve) => {
finish = resolve
})
const observed = vi.spyOn(work, 'then')
const anchorController = new AbortController()
const anchor = waitForPromiseWithSignal(work, anchorController.signal)
for (let index = 0; index < 100; index += 1) {
const controller = new AbortController()
const cancelled = waitForPromiseWithSignal(work, controller.signal)
controller.abort(new Error('cancelled'))
await expect(cancelled).rejects.toThrow('cancelled')
expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0)
}
expect(observed).toHaveBeenCalledOnce()
expect(sharedPromiseSettlementWaiters(work).waiterCount).toBe(1)
finish(42)
await expect(anchor).resolves.toBe(42)
expect(sharedPromiseSettlementWaiters(work).waiterCount).toBe(0)
expect(getEventListeners(anchorController.signal, 'abort')).toHaveLength(0)
})
it('observes shared work only once for callers already aborted', async () => {
const work = new Promise<void>(() => {})
const observed = vi.spyOn(work, 'then')
const signal = AbortSignal.abort(new Error('cancelled'))
for (let index = 0; index < 100; index += 1) {
await expect(waitForPromiseWithSignal(work, signal)).rejects.toThrow('cancelled')
}
expect(observed).toHaveBeenCalledOnce()
expect(sharedPromiseSettlementWaiters(work).waiterCount).toBe(0)
})
it('allows synchronous cancellation before a cached settlement is delivered', async () => {
const work = Promise.resolve(42)
await expect(waitForPromiseWithSignal(work, new AbortController().signal)).resolves.toBe(42)
const controller = new AbortController()
const waiting = waitForPromiseWithSignal(work, controller.signal)
controller.abort(new Error('cancelled'))
await expect(waiting).rejects.toThrow('cancelled')
await expect(waitForPromiseWithSignal(work, new AbortController().signal)).resolves.toBe(42)
})
it('returns work directly when no cancellation was requested', () => {
const work = Promise.resolve(42)
expect(waitForPromiseWithSignal(work)).toBe(work)
})
it('preserves an abort queued between two callers of an already-settled promise', async () => {
const work = Promise.resolve(42)
const first = waitForPromiseWithSignal(work, new AbortController().signal)
const controller = new AbortController()
queueMicrotask(() => controller.abort(new Error('cancelled before this caller could read')))
const second = waitForPromiseWithSignal(work, controller.signal)
await expect(first).resolves.toBe(42)
await expect(second).rejects.toThrow('cancelled before this caller could read')
})
+6 -12
View File
@@ -1,3 +1,5 @@
import { sharedPromiseSettlementWaiters } from './promise-settlement-waiters'
export function abortSignalReason(signal: AbortSignal): Error {
if (signal.reason instanceof Error) {
return signal.reason
@@ -15,17 +17,9 @@ export function waitForPromiseWithSignal<T>(promise: Promise<T>, signal?: AbortS
if (!signal) {
return promise
}
if (signal.aborted) {
// Observe abandoned work so its later rejection stays handled.
void promise.catch(() => undefined)
return Promise.reject(abortSignalReason(signal))
}
return new Promise<T>((resolve, reject) => {
const onAbort = (): void => {
signal.removeEventListener('abort', onAbort)
reject(abortSignalReason(signal))
}
signal.addEventListener('abort', onAbort, { once: true })
void promise.then(resolve, reject).finally(() => signal.removeEventListener('abort', onAbort))
return sharedPromiseSettlementWaiters(promise).wait({
signal,
deferSettled: true,
createAbortError: () => abortSignalReason(signal)
})
}
+28 -18
View File
@@ -1,3 +1,5 @@
import { ownRetainedString } from './own-retained-string'
export const NDJSON_MAX_LINE_BYTES = 16 * 1024 * 1024
const REJECTED_LINE_PREFIX_MAX_BYTES = 64 * 1024
@@ -58,6 +60,11 @@ function boundedUtf8Prefix(value: string, maxBytes: number): string {
return value.slice(0, low)
}
// Why: a strict slice is a V8 SlicedString that pins all of `source`; the whole string needs no copy.
function detachFrom(part: string, source: string): string {
return part.length === source.length ? part : ownRetainedString(part)
}
/** Incremental main-process NDJSON framing with bounded unterminated-line retention. */
export function createIncrementalNdjsonFramer(
onRecord: (record: unknown, line: string) => void,
@@ -100,7 +107,7 @@ export function createIncrementalNdjsonFramer(
prefixBytes += prefix === segment ? segmentBytes : Buffer.byteLength(prefix, 'utf8')
}
const queuePausedCompleteInput = (complete: string): void => {
const queuePausedCompleteInput = (complete: string, source: string): void => {
if (complete.length === 0 || pausedCompleteInputOverflowed) {
return
}
@@ -111,15 +118,15 @@ export function createIncrementalNdjsonFramer(
kind: 'line-too-long',
maxLineBytes: maxQueuedCompleteInputBytes,
observedBytes: pausedCompleteInputBytes + completeBytes,
prefix: boundedUtf8Prefix(complete, REJECTED_LINE_PREFIX_MAX_BYTES)
prefix: ownRetainedString(boundedUtf8Prefix(complete, REJECTED_LINE_PREFIX_MAX_BYTES))
})
return
}
pausedCompleteInput.push(complete)
pausedCompleteInput.push(detachFrom(complete, source))
pausedCompleteInputBytes += completeBytes
}
const retainPendingSuffix = (suffix: string): void => {
const retainPendingSuffix = (suffix: string, source: string): void => {
if (suffix.length === 0) {
pendingInput = null
return
@@ -130,13 +137,13 @@ export function createIncrementalNdjsonFramer(
kind: 'line-too-long',
maxLineBytes: maxPendingInputBytes,
observedBytes: suffixBytes,
prefix: boundedUtf8Prefix(suffix, REJECTED_LINE_PREFIX_MAX_BYTES)
prefix: ownRetainedString(boundedUtf8Prefix(suffix, REJECTED_LINE_PREFIX_MAX_BYTES))
})
pendingInput = null
discardingOversizedLine = true
return
}
pendingInput = suffix
pendingInput = detachFrom(suffix, source)
}
const process = (input: string): void => {
@@ -158,21 +165,23 @@ export function createIncrementalNdjsonFramer(
} else {
const segmentBytes = unbounded ? 0 : Buffer.byteLength(segment, 'utf8')
const nextLineBytes = lineBytes + segmentBytes
if (!unbounded) {
rememberPrefix(segment, segmentBytes)
}
if (nextLineBytes > maxLineBytes) {
rememberPrefix(segment, segmentBytes)
const rejected: NdjsonRejectedRecord = {
kind: 'line-too-long',
maxLineBytes,
observedBytes: nextLineBytes,
prefix: prefixSegments.join('')
prefix: ownRetainedString(prefixSegments.join(''))
}
clearLine()
discardingOversizedLine = !hasNewline
onRejected(rejected)
} else if (!hasNewline) {
lineSegments.push(segment)
const pending = detachFrom(segment, input)
if (!unbounded) {
rememberPrefix(pending, segmentBytes)
}
lineSegments.push(pending)
lineBytes = nextLineBytes
return
} else {
@@ -197,8 +206,8 @@ export function createIncrementalNdjsonFramer(
const newlineIndex = remainder.lastIndexOf('\n')
const complete = newlineIndex === -1 ? '' : remainder.slice(0, newlineIndex + 1)
const suffix = newlineIndex === -1 ? remainder : remainder.slice(newlineIndex + 1)
queuePausedCompleteInput(complete)
retainPendingSuffix(suffix)
queuePausedCompleteInput(complete, input)
retainPendingSuffix(suffix, input)
return
}
}
@@ -223,8 +232,8 @@ export function createIncrementalNdjsonFramer(
const newlineIndex = chunk.lastIndexOf('\n')
const complete = newlineIndex === -1 ? '' : chunk.slice(0, newlineIndex + 1)
const suffix = newlineIndex === -1 ? chunk : chunk.slice(newlineIndex + 1)
queuePausedCompleteInput(complete)
retainPendingSuffix(suffix)
queuePausedCompleteInput(complete, chunk)
retainPendingSuffix(suffix, chunk)
return
}
process(chunk)
@@ -234,8 +243,8 @@ export function createIncrementalNdjsonFramer(
const newlineIndex = combined.lastIndexOf('\n')
const complete = newlineIndex === -1 ? '' : combined.slice(0, newlineIndex + 1)
const suffix = newlineIndex === -1 ? combined : combined.slice(newlineIndex + 1)
queuePausedCompleteInput(complete)
retainPendingSuffix(suffix)
queuePausedCompleteInput(complete, combined)
retainPendingSuffix(suffix, combined)
return
}
process(chunk)
@@ -258,7 +267,8 @@ export function createIncrementalNdjsonFramer(
// behind any newly queued records so it cannot overtake them.
const queuedSuffix = pendingInput
pendingInput = null
retainPendingSuffix(`${queuedSuffix ?? ''}${input}`)
const suffix = `${queuedSuffix ?? ''}${input}`
retainPendingSuffix(suffix, suffix)
} else {
process(input)
}
@@ -0,0 +1,80 @@
import { describe, expect, it } from 'vitest'
import { createIncrementalNdjsonFramer } from './main-process-ndjson-framer'
const CHUNK_CHARS = 1024 * 1024
const READERS = 32
const TAIL = 'tail'.repeat(1024)
function collectHeap(): number {
if (!globalThis.gc) {
throw new Error('Run this regression with --expose-gc')
}
// V8's last regexp input is a separate owner of the record under test.
void /reset/.test('reset')
globalThis.gc()
globalThis.gc()
return process.memoryUsage().heapUsed
}
describe('NDJSON retained chunk memory', () => {
it.each(['tail', 'paused', 'rejected'] as const)(
'releases consumed megabyte records while retaining %s content exactly',
(mode) => {
const held: { finish: () => void; result: () => string }[] = []
const baseline = collectHeap()
for (let index = 0; index < READERS; index += 1) {
let paused = false
let result = ''
const framer = createIncrementalNdjsonFramer(
(record) => {
if (record && typeof record === 'object' && 'tail' in record) {
result = typeof record.tail === 'string' ? record.tail : ''
}
if (mode === 'paused') {
paused = true
}
},
(rejected) => {
if (rejected.kind === 'line-too-long') {
result = rejected.prefix
}
},
{
maxLineBytes: mode === 'rejected' ? 1024 : CHUNK_CHARS * 2,
shouldPause: () => paused
}
)
const payload = String.fromCharCode(65 + (index % 26)).repeat(CHUNK_CHARS)
const chunk = `{"data":"${payload}"}\n${mode === 'paused' ? '{}\n' : ''}${mode === 'rejected' ? '' : `{"tail":"${TAIL}`}`
framer.feed(chunk)
held.push({
finish: () => {
if (mode === 'paused') {
paused = false
framer.resume()
// Delivering the queued record pauses again before the unfinished tail.
paused = false
framer.resume()
}
if (mode !== 'rejected') {
framer.feed('"}\n')
}
},
result: () => result
})
}
const retainedBytes = collectHeap() - baseline
expect(held).toHaveLength(READERS)
expect(retainedBytes).toBeLessThan(8 * 1024 * 1024)
for (const reader of held) {
reader.finish()
if (mode === 'rejected') {
expect(reader.result()).toHaveLength(64 * 1024)
expect(reader.result().startsWith('{"data":"')).toBe(true)
} else {
expect(reader.result()).toBe(TAIL)
}
}
}
)
})
+34 -1
View File
@@ -3,6 +3,7 @@ type PromiseSettlement<T> =
| { status: 'rejected'; error: unknown }
type PromiseSettlementWaiter<T> = {
ready: boolean
resolve: (value: T) => void
reject: (error: unknown) => void
onFulfilled?: (value: T) => void
@@ -13,6 +14,9 @@ type PromiseSettlementWaiter<T> = {
export type PromiseSettlementWaitOptions<T> = {
signal?: AbortSignal
/** Defer delivery a microtask so an abort queued after this wait() still wins.
* Later callers settle in the first caller's reaction turn, not their own. */
deferSettled?: boolean
/** Preserve Promise.race ordering when raw settlement and abort share a turn. */
abortInMicrotask?: boolean
timeoutMs?: number
@@ -47,7 +51,7 @@ export class PromiseSettlementWaiters<T> {
options.onAbandon?.('abort')
return Promise.reject(options.createAbortError?.() ?? createDefaultAbortError())
}
if (this.settlement) {
if (this.settlement && !options.deferSettled) {
return settleWaiterImmediately(this.settlement, options.onFulfilled)
}
return new Promise<T>((resolve, reject) => {
@@ -69,6 +73,7 @@ export class PromiseSettlementWaiters<T> {
}
}
waiter = {
ready: !options.deferSettled,
resolve,
reject,
onFulfilled: options.onFulfilled,
@@ -91,6 +96,16 @@ export class PromiseSettlementWaiters<T> {
if (options.signal?.aborted) {
onAbort()
}
if (options.deferSettled) {
queueMicrotask(() => {
waiter.ready = true
const settlement = this.settlement
if (settlement && this.waiters.delete(waiter)) {
cleanupWaiter(waiter)
settleWaiter(settlement, waiter)
}
})
}
})
}
@@ -100,6 +115,9 @@ export class PromiseSettlementWaiters<T> {
}
this.settlement = settlement
for (const waiter of this.waiters) {
if (!waiter.ready) {
continue
}
this.waiters.delete(waiter)
cleanupWaiter(waiter)
settleWaiter(settlement, waiter)
@@ -108,6 +126,21 @@ export class PromiseSettlementWaiters<T> {
}
}
const sharedWaiters = new WeakMap<Promise<unknown>, PromiseSettlementWaiters<unknown>>()
/** One settlement observer per shared operation; abandoned callers can detach independently. */
export function sharedPromiseSettlementWaiters<T>(
promise: Promise<T>
): PromiseSettlementWaiters<T> {
let waiters = sharedWaiters.get(promise)
if (!waiters) {
waiters = new PromiseSettlementWaiters<unknown>(promise)
sharedWaiters.set(promise, waiters)
}
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the promise key supplies this observer's only settlement, whose value is T.
return waiters as PromiseSettlementWaiters<T>
}
function cleanupWaiter<T>(waiter: PromiseSettlementWaiter<T>): void {
if (waiter.timer) {
clearTimeout(waiter.timer)