timmy-talking-turd/tests/image-concurrency.test.js

// Concurrency ceiling contract.
//
// The decoder runs as a subprocess and each one costs real memory. The service
// runs under MemoryMax=512M, so a burst must be bounded by a small fail-fast
// ceiling rather than queued indefinitely: excess requests are refused
// immediately with a sanitized retry message, and the number of decoder
// children alive at once never exceeds the ceiling.
import test from 'node:test';
import assert from 'node:assert/strict';
import { chmod, mkdir, readFile, readdir, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';

import {
  INGRESS_MAX_CONCURRENCY,
  IngressUnavailableError,
  ingressConcurrencyState,
  validateImageIngress,
} from '../src/image-ingress.js';

const fixture = (name) => fileURLToPath(new URL(`../tests/fixtures/${name}`, import.meta.url));
const clean = await readFile(fixture('ingress-clean.jpg'));
const cleanDataUrl = `data:image/jpeg;base64,${clean.toString('base64')}`;

// An instrumented stand-in interpreter: it records one marker file per spawn
// while it is alive, so the test can observe the true peak child count instead
// of trusting an internal counter.
async function trackingInterpreter(name, holdSeconds = '0.6') {
  const dir = join(tmpdir(), `timmy-ingress-conc-${name}`);
  await rm(dir, { recursive: true, force: true });
  await mkdir(join(dir, 'live'), { recursive: true });
  await mkdir(join(dir, 'seen'), { recursive: true });
  const path = join(dir, 'python3');
  await writeFile(path, [
    '#!/bin/sh',
    `marker="${dir}/live/$$"`,
    `touch "$marker" "${dir}/seen/$$"`,
    `sleep ${holdSeconds}`,
    'rm -f "$marker"',
    // Emit a valid success verdict and write the expected output file.
    'out=""',
    'while [ $# -gt 0 ]; do',
    '  if [ "$1" = "--out" ]; then out="$2"; fi',
    '  shift',
    'done',
    `cp "${fixture('ingress-clean.jpg')}" "$out"`,
    'echo \'{"ok": true, "format": "jpeg", "width": 64, "height": 64, "bytes": 639, "metadataStripped": true}\'',
  ].join('\n'));
  await chmod(path, 0o755);
  return { path, dir };
}

const withPython = async (path, run) => {
  const previous = process.env.TIMMY_PYTHON;
  process.env.TIMMY_PYTHON = path;
  try {
    return await run();
  } finally {
    if (previous === undefined) delete process.env.TIMMY_PYTHON;
    else process.env.TIMMY_PYTHON = previous;
  }
};

test('the concurrency ceiling is small enough to stay inside the service memory budget', () => {
  assert.equal(typeof INGRESS_MAX_CONCURRENCY, 'number');
  assert.ok(INGRESS_MAX_CONCURRENCY >= 1, 'at least one decode must be permitted');
  assert.ok(INGRESS_MAX_CONCURRENCY <= 4,
    `ceiling ${INGRESS_MAX_CONCURRENCY} is too high for a 512 MiB service`);
  // Worst observed single-decoder peak is ~24 MiB; the whole burst must leave
  // ample headroom under MemoryMax=512M alongside the Node process itself.
  assert.ok(INGRESS_MAX_CONCURRENCY * 64 * 1024 * 1024 < 512 * 1024 * 1024,
    'the ceiling must bound peak decoder memory well under 512 MiB');
});

test('a burst beyond the ceiling never runs more decoder children than the ceiling', async () => {
  const { path, dir } = await trackingInterpreter('peak');
  await withPython(path, async () => {
    let peakLive = 0;
    const sampler = setInterval(async () => {
      try {
        const live = await readdir(join(dir, 'live'));
        peakLive = Math.max(peakLive, live.length);
      } catch { /* directory races are fine */ }
    }, 15);
    const burst = await Promise.allSettled(
      Array.from({ length: 24 }, () => validateImageIngress({ consent: true, imageDataUrl: cleanDataUrl })),
    );
    clearInterval(sampler);
    const spawned = (await readdir(join(dir, 'seen'))).length;
    const rejected = burst.filter((r) => r.status === 'rejected');

    assert.ok(peakLive <= INGRESS_MAX_CONCURRENCY,
      `observed ${peakLive} concurrent decoder children, ceiling is ${INGRESS_MAX_CONCURRENCY}`);
    assert.ok(spawned <= INGRESS_MAX_CONCURRENCY,
      `a 24-request burst spawned ${spawned} decoders; only ${INGRESS_MAX_CONCURRENCY} may run and the rest must be refused`);
    assert.ok(rejected.length >= 24 - INGRESS_MAX_CONCURRENCY,
      'excess requests must be refused rather than queued');
  });
});

test('over-ceiling requests fail fast with a sanitized unavailable message', async () => {
  const { path } = await trackingInterpreter('failfast');
  await withPython(path, async () => {
    const started = Date.now();
    const burst = await Promise.allSettled(
      Array.from({ length: 12 }, () => validateImageIngress({ consent: true, imageDataUrl: cleanDataUrl })),
    );
    const rejected = burst.filter((r) => r.status === 'rejected').map((r) => r.reason);
    assert.ok(rejected.length > 0, 'a 12-request burst must refuse some requests');
    for (const error of rejected) {
      assert.ok(error instanceof IngressUnavailableError,
        'an over-capacity refusal is a server-capacity condition, not corrupt client input');
      assert.match(error.message, /temporarily unavailable|slow down|try again/i);
      assert.doesNotMatch(error.message, /corrupt|malformed/i);
      assert.doesNotMatch(error.message, /[A-Za-z0-9+/]{40,}/, 'no payload data in refusals');
      assert.ok(error.message.length < 200);
    }
    // Fail-fast: refusals must not wait for the in-flight decoders to finish.
    assert.ok(Date.now() - started < 5000, 'refusals must be immediate, not queued behind decodes');
  });
});

test('capacity is released so later requests succeed after a burst', async () => {
  const { path } = await trackingInterpreter('release', '0.05');
  await withPython(path, async () => {
    await Promise.allSettled(
      Array.from({ length: 12 }, () => validateImageIngress({ consent: true, imageDataUrl: cleanDataUrl })),
    );
    const state = ingressConcurrencyState();
    assert.equal(state.active, 0, 'every slot must be released after the burst settles');
    const after = await validateImageIngress({ consent: true, imageDataUrl: cleanDataUrl });
    assert.equal(after.format, 'jpeg');
  });
});

test('rejected input releases its slot too', async () => {
  const garbage = await readFile(fixture('ingress-garbage.jpg'));
  for (let i = 0; i < INGRESS_MAX_CONCURRENCY + 3; i += 1) {
    await assert.rejects(() => validateImageIngress({
      consent: true, imageDataUrl: `data:image/jpeg;base64,${garbage.toString('base64')}`,
    }));
  }
  assert.equal(ingressConcurrencyState().active, 0,
    'a rejected request must not leak a concurrency slot');
});