test / src /services /implementations /offsetIndex.service.ts
bipulsardar's picture
fix: Open and close file descriptors per read operation to prevent EIO errors on network mounts
8c3f6a6
Raw History Blame Contribute Delete
5.64 kB
import fs from 'fs';
import { IOffsetIndexService } from '../interfaces/IOffsetIndexService';
import { RowRangeOffset } from '../../types/dataset.types';
import { CONFIG } from '../../config/app.config';
import { logger } from '../../utils/logger';
export class OffsetIndexService implements IOffsetIndexService {
private idxPath: string;
private csvPath: string;
private fileHandle: fs.promises.FileHandle | null = null;
private isLittleEndian: boolean = true;
private endiannessDetected: boolean = false;
private cachedTotalRows: number | null = null;
constructor(customIdxPath?: string, customCsvPath?: string) {
this.idxPath = customIdxPath || CONFIG.ANNOTATIONS_IDX;
this.csvPath = customCsvPath || CONFIG.ANNOTATIONS_CSV;
}
public async isIndexAvailable(): Promise<boolean> {
try {
if (!fs.existsSync(this.idxPath)) {
return false;
}
const stat = await fs.promises.stat(this.idxPath);
return stat.size >= 8;
} catch {
return false;
}
}
private cachedCsvSize: number | null = null;
private async detectEndianness(handle: fs.promises.FileHandle): Promise<void> {
if (this.endiannessDetected) return;
try {
let csvSize = this.cachedCsvSize || 10 * 1024 * 1024 * 1024 * 1024; // Default fallback upper bound (10 TB)
if (this.cachedCsvSize === null && fs.existsSync(this.csvPath)) {
const stat = await fs.promises.stat(this.csvPath);
if (stat.size > 0) {
this.cachedCsvSize = stat.size;
csvSize = stat.size;
}
}
// Read row 1 offset (at byte position 8) to auto-detect endianness
const buf = Buffer.alloc(8);
const { bytesRead } = await handle.read(buf, 0, 8, 8);
if (bytesRead === 8) {
const valLE = Number(buf.readBigUInt64LE(0));
const valBE = Number(buf.readBigUInt64BE(0));
if (valLE > csvSize && valBE <= csvSize) {
this.isLittleEndian = false;
logger.info(`Auto-detected Big-Endian format for index file '${this.idxPath}'`);
} else if (valBE > csvSize && valLE <= csvSize) {
this.isLittleEndian = true;
logger.info(`Auto-detected Little-Endian format for index file '${this.idxPath}'`);
}
}
} catch (err) {
logger.warn(`Failed to auto-detect endianness for '${this.idxPath}', defaulting to Little-Endian:`, err);
} finally {
this.endiannessDetected = true;
}
}
private async getFileHandle(): Promise<fs.promises.FileHandle | null> {
if (this.fileHandle) return this.fileHandle;
try {
if (await this.isIndexAvailable()) {
this.fileHandle = await fs.promises.open(this.idxPath, 'r');
await this.detectEndianness(this.fileHandle);
return this.fileHandle;
}
} catch (err) {
logger.error(`Failed to open index file at ${this.idxPath}:`, err);
}
return null;
}
public async getTotalIndexedRows(): Promise<number> {
if (this.cachedTotalRows !== null) return this.cachedTotalRows;
if (!fs.existsSync(this.idxPath)) {
const errorMsg = `Index file does not exist at '${this.idxPath}'`;
logger.error(errorMsg);
throw new Error(errorMsg);
}
try {
const stat = await fs.promises.stat(this.idxPath);
if (stat.size < 8) {
const errorMsg = `Index file at '${this.idxPath}' is invalid or empty (size: ${stat.size} bytes, minimum required: 8 bytes)`;
logger.error(errorMsg);
throw new Error(errorMsg);
}
// Each uint64 entry is 8 bytes
this.cachedTotalRows = Math.floor(stat.size / 8);
logger.info(`Successfully retrieved ${this.cachedTotalRows} rows from index file '${this.idxPath}'`);
return this.cachedTotalRows;
} catch (err: any) {
const errorMsg = err.message || String(err);
if (!errorMsg.startsWith('Index file')) {
logger.error(`Failed to load index file '${this.idxPath}': ${errorMsg}`);
}
throw new Error(errorMsg.startsWith('Index file') ? errorMsg : `Failed to load index file '${this.idxPath}': ${errorMsg}`);
}
}
public async getOffsetForRow(rowNumber: number): Promise<number> {
if (!fs.existsSync(this.idxPath)) {
return rowNumber * 128;
}
let handle: fs.promises.FileHandle | null = null;
try {
handle = await fs.promises.open(this.idxPath, 'r');
await this.detectEndianness(handle);
const position = rowNumber * 8;
const buf = Buffer.alloc(8);
const { bytesRead } = await handle.read(buf, 0, 8, position);
if (bytesRead < 8) {
return rowNumber * 128;
}
const valBig = this.isLittleEndian ? buf.readBigUInt64LE(0) : buf.readBigUInt64BE(0);
const val = Number(valBig);
if (!Number.isFinite(val) || val < 0) {
return rowNumber * 128;
}
return val;
} catch (err) {
logger.warn(`Error reading offset for row ${rowNumber}:`, err);
return rowNumber * 128;
} finally {
if (handle) {
try { await handle.close(); } catch {}
}
}
}
public async getOffsetsForRange(startRow: number, endRow: number): Promise<RowRangeOffset> {
const startOffset = await this.getOffsetForRow(startRow);
const endOffset = await this.getOffsetForRow(endRow);
const byteLength = Math.max(0, endOffset - startOffset);
return {
startRow,
endRow,
startOffset,
endOffset,
byteLength,
};
}
public async close(): Promise<void> {
if (this.fileHandle) {
await this.fileHandle.close();
this.fileHandle = null;
}
}
}