evomap__evolver/test/syncEngineLoopResilience.test.js

// SyncEngine loop-resilience coverage.
//
// Pre-fix shape (same root cause as the heartbeat #544 / PR #147 bug):
//   setTimeout(async () => {
//     try { await outbound.flush(); } catch { ... }
//     const nextDelay = store.countPending(...) > 0 ? 1_000 : DEFAULT;  // ← outside try
//     this._scheduleOutbound(nextDelay);                                  // ← outside try
//   })
//
// A throw from `store.countPending` (corrupt store file, FS hiccup, locked
// JSONL) escaped the setTimeout callback. Node logged the unhandled
// rejection and `_scheduleOutbound(nextDelay)` was never called. The
// outbound sync loop silently died — `engine._running` stayed true with no
// timer, no signal to the caller. These tests pin the post-fix contract:
// even an exploding `countPending` / `_isIdle` / `flush` must NOT park the
// loop.

const { describe, it, beforeEach, afterEach } = require('node:test');
const assert = require('node:assert/strict');

// Stub the lifecycle/manager AuthError before requiring the engine so the
// engine's `require('../lifecycle/manager')` resolves a minimal shim and we
// don't pull in the full manager (which expects hub creds, env, etc).
const lifecyclePath = require.resolve('../src/proxy/lifecycle/manager');
class AuthError extends Error { constructor(m) { super(m); this.name = 'AuthError'; } }
require.cache[lifecyclePath] = {
  id: lifecyclePath,
  filename: lifecyclePath,
  loaded: true,
  exports: { AuthError },
};

const { SyncEngine } = require('../src/proxy/sync/engine');

// Quiet logger so the deliberate error paths don't spam test output but the
// asserts on what got logged remain straightforward.
function makeQuietLogger() {
  return {
    _errors: [],
    _logs: [],
    log: (...a) => { /* noop */ },
    error: function (...a) { this._errors.push(a.join(' ')); },
    warn: () => {},
  };
}

// Minimal store stub — implementations may throw on demand to exercise the
// resilience paths. countPending defaults to a value > 0 so the loop picks
// the fast cadence and `setTimeout(_, 1_000)` arms quickly.
function makeStore(overrides = {}) {
  return Object.assign({
    countPending: () => 1,
  }, overrides);
}

async function waitFor(predicate, { timeoutMs = 2000, intervalMs = 25 } = {}) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (predicate()) return true;
    await new Promise((r) => setTimeout(r, intervalMs));
  }
  return false;
}

describe('SyncEngine outbound loop resilience', () => {
  let engine;

  afterEach(() => {
    if (engine) try { engine.stop(); } catch (_) {}
    engine = null;
  });

  it('outbound flush throwing does NOT park the loop', async () => {
    let flushCalls = 0;
    const outbound = {
      flush: async () => {
        flushCalls++;
        throw new Error('simulated flush failure');
      },
    };
    const inbound = { pull: async () => ({ received: 0 }), ackDelivered: async () => {} };
    const logger = makeQuietLogger();

    engine = new SyncEngine({
      store: makeStore(),
      hubUrl: 'https://hub.example/test',
      getHeaders: () => ({}),
      logger,
    });
    // Swap in our stubbed senders after construction so the real ones never
    // touch the network.
    engine.outbound = outbound;
    engine.inbound = inbound;

    engine.start();
    // After two flushes the loop must still be alive — i.e. _outTimer is
    // armed and flushCalls keeps incrementing.
    const sawMultipleFlushes = await waitFor(() => flushCalls >= 2, { timeoutMs: 3000 });
    assert.ok(sawMultipleFlushes,
      'flush must keep being called after a throw — loop must NOT silently die. flushCalls=' + flushCalls);
    assert.ok(engine._outTimer, 'outbound timer must remain armed after a flush throw');
  });

  it('store.countPending throwing does NOT park the loop (the #544 pattern in sync engine)', async () => {
    // After a successful flush the post-tick path calls store.countPending
    // to pick the next cadence (1s if pending > 0, else 5s). Pre-fix this
    // call was OUTSIDE the try/catch, so a throw here killed the loop.
    // Post-fix: the throw is caught and the loop falls back to the default
    // 5s cadence. Asserting the timer is re-armed after the first tick is
    // the property we actually need.
    let flushCalls = 0;
    let countPendingCalls = 0;
    const outbound = {
      flush: async () => {
        flushCalls++;
        return { sent: 0 };
      },
    };
    const inbound = { pull: async () => ({ received: 0 }), ackDelivered: async () => {} };
    const store = {
      countPending: () => {
        countPendingCalls++;
        throw new Error('simulated store corruption (countPending exploded)');
      },
    };
    const logger = makeQuietLogger();

    engine = new SyncEngine({
      store,
      hubUrl: 'https://hub.example/test',
      getHeaders: () => ({}),
      logger,
    });
    engine.outbound = outbound;
    engine.inbound = inbound;
    const initialOutTimer = engine._outTimer;

    engine.start();

    const sawFlushAndCount = await waitFor(
      () => flushCalls >= 1 && countPendingCalls >= 1,
      { timeoutMs: 3000 },
    );
    assert.ok(sawFlushAndCount,
      'flush + countPending must each fire at least once within 3s. flushCalls=' +
      flushCalls + ', countPendingCalls=' + countPendingCalls);
    // Let the finally block run one event-loop turn to re-arm the timer.
    await new Promise((r) => setTimeout(r, 50));
    assert.ok(engine._outTimer,
      'outbound timer must be RE-ARMED after countPending throws (the #544-class fix)');
    assert.notEqual(engine._outTimer, initialOutTimer,
      'a new timer must be installed (the rescheduled one), not the original');

    // Logger must have recorded the countPending failure (non-fatal).
    const sawCountPendingError = logger._errors.some((e) => e.includes('countPending'));
    assert.ok(sawCountPendingError,
      'countPending failure must be logged (non-fatal) — errors=' + JSON.stringify(logger._errors));
  });
});

describe('SyncEngine inbound loop resilience', () => {
  let engine;

  afterEach(() => {
    if (engine) try { engine.stop(); } catch (_) {}
    engine = null;
  });

  it('inbound pull throwing does NOT park the loop', async () => {
    // Inbound cadence is DEFAULT_POLL_INTERVAL_ACTIVE = 10s, so waiting for
    // a second call would make the test very slow. Instead: wait for the
    // first pull (1s after start()), wait one event-loop turn for the
    // setTimeout callback to fully resolve through `finally`, then assert
    // the next timer is armed. That's the property we actually care about
    // — the loop didn't die.
    let pullCalls = 0;
    const inbound = {
      pull: async () => {
        pullCalls++;
        throw new Error('simulated inbound pull failure');
      },
      ackDelivered: async () => {},
    };
    const outbound = { flush: async () => ({ sent: 0 }) };
    const logger = makeQuietLogger();

    engine = new SyncEngine({
      store: makeStore(),
      hubUrl: 'https://hub.example/test',
      getHeaders: () => ({}),
      logger,
    });
    engine.outbound = outbound;
    engine.inbound = inbound;
    const initialInTimer = engine._inTimer;

    engine.start();

    const sawPull = await waitFor(() => pullCalls >= 1, { timeoutMs: 3000 });
    assert.ok(sawPull, 'first pull must fire within 3s of start(). pullCalls=' + pullCalls);
    // The throw resolves the setTimeout callback; let the finally block
    // run one event-loop turn so it can re-arm the timer.
    await new Promise((r) => setTimeout(r, 50));
    assert.ok(engine._inTimer, 'inbound timer must be RE-ARMED after a pull throw');
    assert.notEqual(engine._inTimer, initialInTimer,
      'a new timer must have been created (the rescheduled one), not the original');
  });

  it('onInboundReceived callback throwing does NOT park the loop', async () => {
    // Same shape as above: assert the timer is re-armed after one tick
    // rather than waiting for a full 10s second cycle.
    let pullCalls = 0;
    let callbackCalls = 0;
    const inbound = {
      pull: async () => {
        pullCalls++;
        return { received: 1 };
      },
      ackDelivered: async () => {},
    };
    const outbound = { flush: async () => ({ sent: 0 }) };
    const logger = makeQuietLogger();

    engine = new SyncEngine({
      store: makeStore(),
      hubUrl: 'https://hub.example/test',
      getHeaders: () => ({}),
      logger,
      onInboundReceived: () => {
        callbackCalls++;
        throw new Error('user callback exploded');
      },
    });
    engine.outbound = outbound;
    engine.inbound = inbound;

    engine.start();

    const sawPullAndCallback = await waitFor(() => pullCalls >= 1 && callbackCalls >= 1, { timeoutMs: 3000 });
    assert.ok(sawPullAndCallback,
      'first pull + callback must fire within 3s. pullCalls=' + pullCalls + ', callbackCalls=' + callbackCalls);
    await new Promise((r) => setTimeout(r, 50));
    assert.ok(engine._inTimer,
      'inbound timer must be re-armed even when the user callback throws');
  });
});

describe('SyncEngine stop() still wins over the resilience layer', () => {
  it('stop() prevents the reschedule even when called mid-await of the in-flight tick', async () => {
    // Bugbot PR #158: the previous shape of this test waited 30ms then
    // called stop() — but the first timer is armed at 500ms, so stop()
    // ran BEFORE the timer fired and `clearTimeout` cancelled it. The
    // setTimeout callback (which holds the `if (this._running)` guard we
    // actually want to validate) never ran. The test passed even if the
    // guard was removed.
    //
    // The contract the resilience layer must uphold: when stop() runs
    // WHILE a tick is mid-await, the in-flight flush still resolves
    // naturally, but `finally`'s `if (this._running)` gate prevents the
    // next timer from being armed. No further flushes happen.
    let flushStarted = 0;
    let flushFinished = 0;
    const outbound = {
      flush: async () => {
        flushStarted++;
        // Long enough that we can definitely call stop() between
        // flushStarted++ and the resolve below.
        await new Promise((r) => setTimeout(r, 300));
        flushFinished++;
        return { sent: 0 };
      },
    };
    const inbound = { pull: async () => ({ received: 0 }), ackDelivered: async () => {} };
    const logger = makeQuietLogger();

    const engine = new SyncEngine({
      store: makeStore(),
      hubUrl: 'https://hub.example/test',
      getHeaders: () => ({}),
      logger,
    });
    engine.outbound = outbound;
    engine.inbound = inbound;

    engine.start();
    // Wait until the FIRST timer has fired and flush has started (~500ms).
    const flushIsRunning = await waitFor(() => flushStarted >= 1, { timeoutMs: 2000 });
    assert.ok(flushIsRunning, 'first flush must start within 2s of engine.start()');
    // Sanity check: we should be MID-AWAIT (started but not finished yet),
    // otherwise the test is racy and we'd be back to the pre-fix shape
    // where stop() just cancels a not-yet-fired timer.
    assert.equal(flushFinished, 0,
      'flush must still be mid-await when stop() runs — otherwise the test ' +
      'is not exercising the finally-guard path. flushStarted=' + flushStarted);

    engine.stop();
    // Let the in-flight flush resolve through `finally`. The guard is
    // what we are testing: stop() flipped _running=false, so finally
    // must NOT reschedule.
    await new Promise((r) => setTimeout(r, 600));
    assert.equal(flushFinished, 1,
      'in-flight flush must still resolve naturally after stop()');
    assert.equal(engine._outTimer, null,
      'outbound timer must NOT be re-armed by the finally block when stop() flipped _running=false');
    assert.equal(engine._inTimer, null,
      'inbound timer must be cleared by stop() too');

    // Wait beyond the default 5s cadence to prove no rogue timer is
    // hiding. If the finally guard were broken, a second flush would
    // fire here.
    const startedAfterStop = flushStarted;
    await new Promise((r) => setTimeout(r, 1000));
    assert.equal(flushStarted, startedAfterStop,
      'no further flushes after stop() — got ' + flushStarted + ' vs at-stop=' + startedAfterStop);
  });
});