refactor: implement atomic state file updates and improve error handl…

indexer/index.mjs

@@ -1,4 +1,4 @@
-import { appendFileSync, existsSync, readFileSync, writeFileSync } from "node:fs";
+import { appendFileSync, existsSync, readFileSync, renameSync, writeFileSync } from "node:fs";
 import { rpc, scValToNative } from "@stellar/stellar-sdk";
 import { withRateLimitBackoff } from "./rpc-backoff.mjs";
 
@@ -31,20 +31,32 @@ if (!config.ok) {
 
 const { RPC_URL, CONTRACT_ID } = config.value;
 const OUT = process.env.OUT ?? "events.ndjson";
-const STATE = process.env.STATE ?? "state.json";
 const POLL_MS = Number(process.env.POLL_MS ?? 10_000);
 const BACKOFF_INITIAL_MS = Number(process.env.BACKOFF_INITIAL_MS ?? 1_000);
 const BACKOFF_MAX_MS = Number(process.env.BACKOFF_MAX_MS ?? 60_000);
 
 const server = new rpc.Server(RPC_URL);
 
-function loadCursor() {
-  if (!existsSync(STATE)) return null;
-  return JSON.parse(readFileSync(STATE, "utf8")).cursor ?? null;
+function stateFile() {
+  return process.env.STATE ?? "state.json";
 }
 
-function saveCursor(cursor) {
-  writeFileSync(STATE, JSON.stringify({ cursor }));
+export function loadCursor() {
+  const file = stateFile();
+  if (!existsSync(file)) return null;
+  try {
+    return JSON.parse(readFileSync(file, "utf8")).cursor ?? null;
+  } catch {
+    console.warn(`[indexer] state file "${file}" is corrupt or empty; starting from default window`);
+    return null;
+  }
+}
+
+export function saveCursor(cursor) {
+  const file = stateFile();
+  const tmp = `${file}.tmp`;
+  writeFileSync(tmp, JSON.stringify({ cursor }));
+  renameSync(tmp, file);
 }
 
 function decode(ev) {
@@ -173,6 +185,8 @@ async function poll() {
   }
 }
 
-console.log(`indexing ${CONTRACT_ID} from ${RPC_URL} every ${POLL_MS}ms`);
-await poll();
-intervalId = setInterval(() => poll().catch((e) => console.error(e.message ?? e)), POLL_MS);
+if (import.meta.filename === process.argv[1]) {
+  console.log(`indexing ${CONTRACT_ID} from ${RPC_URL} every ${POLL_MS}ms`);
+  await poll();
+  intervalId = setInterval(() => poll().catch((e) => console.error(e.message ?? e)), POLL_MS);
+}

indexer/index.test.mjs

@@ -1,6 +1,14 @@
 import test from 'node:test';
 import assert from 'node:assert/strict';
+import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
 import { validateConfig } from './index.mjs';
+import { loadCursor, saveCursor } from './index.mjs';
+
+// ---------------------------------------------------------------------------
+// validateConfig
+// ---------------------------------------------------------------------------
 
 test('validateConfig rejects missing required env values', () => {
   const result = validateConfig({
@@ -23,3 +31,92 @@ test('validateConfig accepts populated env values', () => {
   assert.equal(result.value.CONTRACT_ID, 'CC123');
   assert.equal(result.value.RPC_URL, 'https://example.com');
 });
+
+// ---------------------------------------------------------------------------
+// loadCursor / saveCursor
+// ---------------------------------------------------------------------------
+
+/**
+ * Create a fresh temporary directory for each test and override the STATE
+ * environment variable so the functions use our isolated file.
+ */
+function withTempState(fn) {
+  return async () => {
+    const dir = mkdtempSync(join(tmpdir(), 'tributary-test-'));
+    const stateFile = join(dir, 'state.json');
+    const original = process.env.STATE;
+    process.env.STATE = stateFile;
+    try {
+      await fn({ dir, stateFile });
+    } finally {
+      process.env.STATE = original ?? undefined;
+      rmSync(dir, { recursive: true, force: true });
+    }
+  };
+}
+
+test('loadCursor returns null when state file does not exist', withTempState(async ({ stateFile }) => {
+  assert.equal(existsSync(stateFile), false, 'precondition: file must not exist');
+  // loadCursor/saveCursor read STATE from process.env at call time via the
+  // module-level binding, so we call them after setting the env var.
+  const cursor = loadCursor();
+  assert.equal(cursor, null);
+}));
+
+test('loadCursor returns null and warns on empty state file', withTempState(async ({ stateFile }) => {
+  writeFileSync(stateFile, '');
+
+  const warnings = [];
+  const origWarn = console.warn;
+  console.warn = (...args) => warnings.push(args.join(' '));
+
+  const cursor = loadCursor();
+
+  console.warn = origWarn;
+
+  assert.equal(cursor, null);
+  assert.ok(warnings.length > 0, 'should have logged a warning');
+  assert.ok(warnings[0].includes('corrupt or empty'), `unexpected warning: ${warnings[0]}`);
+}));
+
+test('loadCursor returns null and warns on corrupt state file', withTempState(async ({ stateFile }) => {
+  writeFileSync(stateFile, '{bad json{{');
+
+  const warnings = [];
+  const origWarn = console.warn;
+  console.warn = (...args) => warnings.push(args.join(' '));
+
+  const cursor = loadCursor();
+
+  console.warn = origWarn;
+
+  assert.equal(cursor, null);
+  assert.ok(warnings.length > 0, 'should have logged a warning');
+}));
+
+test('saveCursor writes atomically and loadCursor reads back the cursor', withTempState(async ({ dir, stateFile }) => {
+  const testCursor = '1234567890-1';
+
+  saveCursor(testCursor);
+
+  // The .tmp file must not be left behind after a successful save.
+  assert.equal(existsSync(`${stateFile}.tmp`), false, '.tmp file should be cleaned up');
+
+  // The state file must exist and contain the correct cursor.
+  assert.ok(existsSync(stateFile), 'state.json must exist after saveCursor');
+
+  const raw = JSON.parse(readFileSync(stateFile, 'utf8'));
+  assert.equal(raw.cursor, testCursor);
+
+  const loaded = loadCursor();
+  assert.equal(loaded, testCursor);
+}));
+
+test('saveCursor overwrites previous cursor atomically', withTempState(async ({ stateFile }) => {
+  saveCursor('cursor-v1');
+  saveCursor('cursor-v2');
+
+  const loaded = loadCursor();
+  assert.equal(loaded, 'cursor-v2');
+  assert.equal(existsSync(`${stateFile}.tmp`), false);
+}));