machineGroupControl/test/basic/demandDispatcher.basic.test.js

'use strict';

const test = require('node:test');
const assert = require('node:assert/strict');

const DemandDispatcher = require('../../src/dispatch/demandDispatcher.js');

const silentLogger = { warn() {}, error() {}, debug() {}, info() {} };

// Helper: a manually-resolvable promise so we can pin a dispatch in flight.
function deferred() {
    let resolve;
    let reject;
    const promise = new Promise((res, rej) => { resolve = res; reject = rej; });
    return { promise, resolve, reject };
}

test('fire(50) triggers runFn with 50', async () => {
    const calls = [];
    const dispatcher = new DemandDispatcher(
        { logger: silentLogger },
        async (demand) => { calls.push(demand); },
    );
    dispatcher.fire(50);
    await dispatcher.drain();
    assert.deepEqual(calls, [50]);
});

test('two fires back-to-back during in-flight — only the second runs after first settles', async () => {
    const calls = [];
    const gates = [deferred()];
    const dispatcher = new DemandDispatcher(
        { logger: silentLogger },
        async (demand) => {
            calls.push(demand);
            await gates[0].promise;
        },
    );

    dispatcher.fire(10);
    // first invocation is now in flight (after a microtask)
    await Promise.resolve();
    await Promise.resolve();
    dispatcher.fire(20);
    // 20 should be pending, not yet run.
    assert.deepEqual(calls, [10]);
    gates[0].resolve();
    await dispatcher.drain();
    assert.deepEqual(calls, [10, 20]);
});

test('three rapid fires — only first + last run; middle dropped', async () => {
    const calls = [];
    const gate = deferred();
    const dispatcher = new DemandDispatcher(
        { logger: silentLogger },
        async (demand) => {
            calls.push(demand);
            if (calls.length === 1) await gate.promise;
        },
    );

    dispatcher.fire(1);
    await Promise.resolve();
    await Promise.resolve();
    dispatcher.fire(2);
    dispatcher.fire(3);   // overwrites the pending 2

    assert.deepEqual(calls, [1]);
    gate.resolve();
    await dispatcher.drain();
    assert.deepEqual(calls, [1, 3]);
});

test('drain() resolves only when idle', async () => {
    const gate = deferred();
    let runs = 0;
    const dispatcher = new DemandDispatcher(
        { logger: silentLogger },
        async () => { runs++; await gate.promise; },
    );

    // drain() on an idle gate resolves immediately.
    await dispatcher.drain();

    dispatcher.fire('a');
    let drained = false;
    const drainPromise = dispatcher.drain().then(() => { drained = true; });
    // Let a few microtasks run — drain must NOT be resolved while in flight.
    for (let i = 0; i < 5; i++) await Promise.resolve();
    assert.equal(drained, false);
    assert.equal(runs, 1);
    gate.resolve();
    await drainPromise;
    assert.equal(drained, true);
});

test('error in runFn does not deadlock; subsequent fire still works', async () => {
    const calls = [];
    const dispatcher = new DemandDispatcher(
        { logger: silentLogger },
        async (demand) => {
            calls.push(demand);
            if (demand === 'boom') throw new Error('boom');
        },
    );
    dispatcher.fire('boom');
    await dispatcher.drain();
    dispatcher.fire('ok');
    await dispatcher.drain();
    assert.deepEqual(calls, ['boom', 'ok']);
});

test('inFlight getter reports correctly', async () => {
    const gate = deferred();
    const dispatcher = new DemandDispatcher(
        { logger: silentLogger },
        async () => { await gate.promise; },
    );
    assert.equal(dispatcher.inFlight, false);
    dispatcher.fire(1);
    // Microtask scheduling — gate flips to inFlight after one tick.
    await Promise.resolve();
    assert.equal(dispatcher.inFlight, true);
    gate.resolve();
    await dispatcher.drain();
    assert.equal(dispatcher.inFlight, false);
});

test('runFn receives the ctx supplied at construction', async () => {
    const seen = [];
    const ctx = { logger: silentLogger, marker: 'mgc-A' };
    const dispatcher = new DemandDispatcher(
        ctx,
        async (demand, runCtx) => { seen.push({ demand, marker: runCtx.marker }); },
    );
    dispatcher.fire(42);
    await dispatcher.drain();
    assert.deepEqual(seen, [{ demand: 42, marker: 'mgc-A' }]);
});