scriptoriumadmin/GitNexus
1
0
Fork 0
mirror of https://github.com/abhigyanpatwari/GitNexus.git synced 2026-10-04 02:31:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

refactor(lbug): extract SyncCsvWriter into a shared module

Browse source 
`PdgEmitSink` (#2202) declared `SyncCsvWriter` as a private, non-exported
class. The structural streaming sink for #2680 needs the same buffered
sync-write + poison/openFailure IO discipline, and importing it is not
possible while it is module-private — so the alternative was copying ~90
lines of it.

Extract the class (and the chunk-rows default it uses) into
`sync-csv-writer.ts` and have `PdgEmitSink` import it.
`DEFAULT_PDG_EMIT_CHUNK_ROWS` stays exported as an alias so no existing
caller changes.

Pure refactor: no behaviour change. pdg-emit-sink.ts 396 -> 302 lines;
tsc clean; the 23 existing #2202 tests pass unchanged.

Refs #2680
This commit is contained in:
Gergo Magyar 2026-07-24 21:25:39 +00:00
parent d3d4fa31bb
commit f69e22ad9f
2 changed files with 114 additions and 97 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

101
gitnexus/src/core/lbug/pdg-emit-sink.ts
View file

@ -54,6 +54,7 @@ import {
buildRelRow,
} from './csv-generator.js';
import { getNodeLabel } from './rel-pair-routing.js';
import { DEFAULT_EMIT_CHUNK_ROWS, SyncCsvWriter } from './sync-csv-writer.js';
import { NODE_TABLES, type NodeTableName } from './schema.js';
/**
@ -73,103 +74,9 @@ const PDG_EDGE_TYPES: ReadonlySet<RelationshipType> = new Set<RelationshipType>(
]);
/** Default streamed-write buffer (rows). Matches the whole-graph emit's
* `FLUSH_EVERY` order of magnitude; overridable via `GITNEXUS_PDG_EMIT_CHUNK_SIZE`. */
export const DEFAULT_PDG_EMIT_CHUNK_ROWS = 500;
/**
* Synchronous buffered CSV writer. Buffers up to `chunkRows` rows, then issues
* one `fs.writeSync` straight to the OS (no in-process stream buffer). Header
* is written into the buffer at construction and is NOT counted in `rows`
* (matching `BufferedCSVWriter` semantics, so manifest row counts line up).
*/
class SyncCsvWriter {
private fd: number;
private buf: string[] = [];
private readonly chunkRows: number;
rows = 0;
/**
* First IO error this writer hit (a `fs.writeSync` short-write loop throwing
* on e.g. disk-full). Once poisoned the writer refuses further rows and
* skips its final flush; the sink surfaces it from {@link PdgEmitSink.finalize}
* so a truncated CSV is never handed to the bulk COPY (#2202 review #4). A
* streamed-write failure is an IO fault, not the CFG-logic error that the
* emit loop's per-file try/catch is built to swallow — poisoning routes it
* past that catch to a loud failure.
*/
poison: unknown | undefined = undefined;
constructor(
readonly csvPath: string,
header: string,
chunkRows: number,
) {
// Guard a 0/negative buffer: the flush modulo would never fire and `buf`
// would grow unbounded, defeating the whole point of streaming.
this.chunkRows = Math.max(1, chunkRows);
// Exclusive create (O_EXCL): the streamed-CSV dir is wiped + recreated fresh
// by the PdgEmitSink constructor before any writer opens a file, so the path
// never pre-exists — 'wx' both matches that invariant and refuses to follow
// a pre-planted symlink at the path (CWE-377 / CodeQL js/insecure-temporary-file).
this.fd = fs.openSync(csvPath, 'wx');
this.buf.push(header);
}
addRow(row: string): void {
// A poisoned writer is dead — stop buffering so memory can't grow on a
// writer whose fd is already in a bad state; finalize will report the fault.
if (this.poison !== undefined) return;
this.buf.push(row);
this.rows++;
// Flush on DATA-row count, not buffer length: the header occupies buf[0]
// until the first flush, so a `buf.length >= chunkRows` test would fire one
// row early on the first chunk. Counting rows makes every flush exactly
// `chunkRows` rows.
if (this.rows % this.chunkRows === 0) this.flushOrPoison();
}
/** Flush, recording (and re-throwing) any IO error as poison. Re-throwing
* lets the immediate caller log the per-file failure; the persisted `poison`
* is the backstop that makes finalize fail loudly even when that throw is
* swallowed by the emit loop's CFG try/catch. */
private flushOrPoison(): void {
try {
this.flush();
} catch (e) {
this.poison ??= e;
throw e;
}
}
private flush(): void {
if (this.buf.length === 0) return;
const data = Buffer.from(this.buf.join('\n') + '\n', 'utf8');
// fs.writeSync can return a short byte count; loop until the whole buffer
// lands so a partial write never truncates a CSV row mid-field.
let offset = 0;
while (offset < data.length) {
offset += fs.writeSync(this.fd, data, offset, data.length - offset);
}
this.buf.length = 0;
}
/** Flush remaining rows (unless already poisoned) and close the fd. Never
* throws: a final-flush IO error is recorded as poison and the fd is still
* closed, so a write error neither leaks an fd nor escapes here — the sink
* reads {@link poison} after closing every writer and fails loudly then. */
close(): void {
try {
if (this.poison === undefined) this.flush();
} catch (e) {
this.poison ??= e;
} finally {
try {
fs.closeSync(this.fd);
} catch {
/* fd may already be invalid after an IO fault — nothing to recover */
}
}
}
}
* `FLUSH_EVERY` order of magnitude; overridable via `GITNEXUS_PDG_EMIT_CHUNK_SIZE`.
* Aliases the shared default in `sync-csv-writer.ts` (#2680 extraction). */
export const DEFAULT_PDG_EMIT_CHUNK_ROWS = DEFAULT_EMIT_CHUNK_ROWS;
/**
* COPY manifest produced by {@link PdgEmitSink.finalize}. Shaped to merge

110
gitnexus/src/core/lbug/sync-csv-writer.ts Normal file
View file

@ -0,0 +1,110 @@
/**
* Synchronous buffered CSV writer, shared by the streaming emit sinks.
*
* Extracted verbatim from `pdg-emit-sink.ts` (issue #2202) so the structural
* `GraphEmitSink` (#2680) reuses the same buffering and IO-fault discipline
* instead of duplicating ~90 lines of it. No behaviour change: `PdgEmitSink`
* imports this class and is otherwise untouched.
*
* Why synchronous? The emit loops these sinks sit under are synchronous — there
* is no `await` point to drain an async stream, so a `WriteStream` would
* accumulate unwritten chunks in process memory across millions of rows,
* defeating the RSS bound this exists to provide. `fs.writeSync` goes straight
* to the OS; resident memory is bounded to one `chunkRows` buffer. This mirrors
* the sync-shard pattern in `storage/parsedfile-store.ts`.
*/
import fs from 'fs';
/** Default streamed-write buffer (rows), shared by both sinks. */
export const DEFAULT_EMIT_CHUNK_ROWS = 500;
export class SyncCsvWriter {
private fd: number;
private buf: string[] = [];
private readonly chunkRows: number;
rows = 0;
/**
* First IO error this writer hit (a `fs.writeSync` short-write loop throwing
* on e.g. disk-full). Once poisoned the writer refuses further rows and
* skips its final flush; the owning sink surfaces it from its `finalize()`
* so a truncated CSV is never handed to the bulk COPY (#2202 review #4). A
* streamed-write failure is an IO fault, not the logic error that the emit
* loops' per-file try/catch is built to swallow — poisoning routes it past
* that catch to a loud failure.
*/
poison: unknown | undefined = undefined;
constructor(
readonly csvPath: string,
header: string,
chunkRows: number,
) {
// Guard a 0/negative buffer: the flush modulo would never fire and `buf`
// would grow unbounded, defeating the whole point of streaming.
this.chunkRows = Math.max(1, chunkRows);
// Exclusive create (O_EXCL): the streamed-CSV dir is wiped + recreated fresh
// by the owning sink's constructor before any writer opens a file, so the
// path never pre-exists — 'wx' both matches that invariant and refuses to
// follow a pre-planted symlink at the path (CWE-377 / CodeQL
// js/insecure-temporary-file).
this.fd = fs.openSync(csvPath, 'wx');
this.buf.push(header);
}
addRow(row: string): void {
// A poisoned writer is dead — stop buffering so memory can't grow on a
// writer whose fd is already in a bad state; finalize will report the fault.
if (this.poison !== undefined) return;
this.buf.push(row);
this.rows++;
// Flush on DATA-row count, not buffer length: the header occupies buf[0]
// until the first flush, so a `buf.length >= chunkRows` test would fire one
// row early on the first chunk. Counting rows makes every flush exactly
// `chunkRows` rows.
if (this.rows % this.chunkRows === 0) this.flushOrPoison();
}
/** Flush, recording (and re-throwing) any IO error as poison. Re-throwing
* lets the immediate caller log the per-file failure; the persisted `poison`
* is the backstop that makes finalize fail loudly even when that throw is
* swallowed by an emit loop's try/catch. */
private flushOrPoison(): void {
try {
this.flush();
} catch (e) {
this.poison ??= e;
throw e;
}
}
private flush(): void {
if (this.buf.length === 0) return;
const data = Buffer.from(this.buf.join('\n') + '\n', 'utf8');
// fs.writeSync can return a short byte count; loop until the whole buffer
// lands so a partial write never truncates a CSV row mid-field.
let offset = 0;
while (offset < data.length) {
offset += fs.writeSync(this.fd, data, offset, data.length - offset);
}
this.buf.length = 0;
}
/** Flush remaining rows (unless already poisoned) and close the fd. Never
* throws: a final-flush IO error is recorded as poison and the fd is still
* closed, so a write error neither leaks an fd nor escapes here — the owning
* sink reads {@link poison} after closing every writer and fails loudly then. */
close(): void {
try {
if (this.poison === undefined) this.flush();
} catch (e) {
this.poison ??= e;
} finally {
try {
fs.closeSync(this.fd);
} catch {
/* fd may already be invalid after an IO fault — nothing to recover */
}
}
}
}