016 — A partial result and a complete one are the same value

{
  "name": "016-partial-and-complete",
  "what_it_is": "A gather step over three sources where one of them hands over real records and then breaks, and where the way the results are combined is itself a declared unit.",
  "the_structural_decision": "A partial result and a complete one are the same value once the records are in one array. So the outcome cannot travel beside the records — it has to travel WITH them, which means a record carries which unit produced it. And the combination policy is a declaring unit: it says what it does with work that already succeeded.",
  "why_this_one": "Example 015 named four outcomes and wrote partial failure into its own limitations as a fifth it did not model. This is that limitation gone, and it turned out not to be a fifth label but a different axis.",
  "status": "candidate",

  "outcomes": {
    "worked": "yielded records and finished",
    "empty": "yielded nothing and finished — a legitimate quiet day",
    "partial": "yielded records and then stopped early — the fifth, and the one that survives aggregation unmarked",
    "failed": "was called and yielded nothing before stopping",
    "absent": "not loaded at all — what the island test produces"
  },

  "who_says_it_finished": "Not the source. The collector drives the iterator and observes where it stopped, so `finished` is not a field a unit can get wrong. What a unit still declares is what it can fail WITH (改良點 13).",

  "sources": {
    "full-batch":    {"yields": 3, "can_fail_with": ["unreadable-page"]},
    "short-batch":   {"yields": 2, "can_fail_with": ["stale-cursor"], "note": "the control — it yields the SAME number of records as breaks-midway and finishes, so a count cannot separate them and the outcome field can"},
    "breaks-midway": {"yields": 2, "can_fail_with": ["connection-reset", "timeout"], "note": "then throws"}
  },

  "combiners": {
    "all-or-nothing": {"keeps_partial_work": false, "upstream": "Promise.all"},
    "settle-each":    {"keeps_partial_work": true,  "upstream": "Promise.allSettled"}
  },

  "the_finding": "Under all-or-nothing, REMOVING the broken source gives more than keeping it: 5 records against 0. That inverts example 015, where removing a source and breaking it produced the same total. Archaeology 016 measures the same inversion in Promise.all — three of four members fulfilled, every one of them ran to completion, and none of their values is reachable from the rejection.",

  "sets": {
    "FMS": "this file: the five outcomes, what each source can fail with, what each combiner does with partial work, and the units map",
    "SCL": "which combiner this deployment runs and whether a partial run is fatal here",
    "SMS": "loading, driving each source, observing where it stopped, and refusing a record with no origin",
    "TMS": "one file per source and one per combiner — each declares itself and reaches no sibling",
    "DMS": "the per-source outcome, what the combiner kept or discarded, and how many kept records came from a source that did not finish"
  },

  "units": {
    "TMS/sources": ["breaks_midway.mjs", "full_batch.mjs", "short_batch.mjs"],
    "TMS/combiners": ["all_or_nothing.mjs", "settle_each.mjs"]
  },

  "non_goals": [
    "Retries, resumption or cursors. Restarting a source that broke at record two is a different problem and this one has to be nameable first.",
    "Deciding which combiner is right. all-or-nothing is correct for a nightly index and wrong for a dashboard; SCL picks one and the example takes no position.",
    "Detecting a source that swallows its own failure internally and returns early. Section 7 of the island test asserts that such a source is indistinguishable from short-batch, so the limit stays measured rather than assumed."
  ]
}

{
  "deployment": "nightly-index-build",
  "combiner": "all-or-nothing",
  "on_partial": "fatal",
  "why": "A nightly search index that publishes a partial crawl serves an index that looks complete and is not. Refusing the whole batch is defensible HERE and is not a general rule — a dashboard would pick settle-each and mark the rows. The example takes no position on which is right; SCL is where the position lives.",
  "what_this_costs": "Every record from every source that did finish is discarded, and the work that produced them ran to completion regardless. --compare prints the number."
}

// Deployment policy. Which combiner this deployment runs, and what a run that
// did not finish means here.
import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";

const here = path.dirname(fileURLToPath(import.meta.url));
export const POLICY = JSON.parse(fs.readFileSync(path.join(here, "policy.json"), "utf8"));

export const combinerName = () => POLICY.combiner;
export const isFatal = () => POLICY.on_partial === "fatal";
export const describe = () =>
  `${POLICY.deployment}: combine with ${POLICY.combiner}, a partial run is ${POLICY.on_partial}`;

// Load the sources and the combiners, drive each source, and classify.
//
// The rule this example adds to example 015's four outcomes: `partial` is not
// a fifth label a unit hands in. A source that throws after yielding cannot
// report that it finished, because it is not the one that says so — the
// collector drives the iterator and observes where it stopped.
//
// The second rule: every record carries `from`. A record that does not is
// refused, because once records are in one array a partial batch and a
// complete batch are the same value.
import * as breaksMidway from "../TMS/sources/breaks_midway.mjs";
import * as fullBatch from "../TMS/sources/full_batch.mjs";
import * as shortBatch from "../TMS/sources/short_batch.mjs";
import * as allOrNothing from "../TMS/combiners/all_or_nothing.mjs";
import * as settleEach from "../TMS/combiners/settle_each.mjs";

export const OUTCOMES = ["worked", "empty", "partial", "failed", "absent"];

export function load(extraSources = [], extraCombiners = []) {
  const sources = {};
  const combiners = {};
  const problems = [];

  for (const module of [fullBatch, shortBatch, breaksMidway, ...extraSources]) {
    for (const attribute of ["NAME", "CAN_FAIL_WITH", "records"]) {
      if (module[attribute] === undefined) problems.push(`a source does not declare ${attribute}`);
    }
    if (!module.CAN_FAIL_WITH?.length) {
      problems.push(`${module.NAME}: CAN_FAIL_WITH is empty - a unit that cannot say what a bad ` +
        `day looks like cannot be reported as degraded`);
    }
    sources[module.NAME] = module;
  }

  for (const module of [allOrNothing, settleEach, ...extraCombiners]) {
    if (typeof module.KEEPS_PARTIAL_WORK !== "boolean") {
      problems.push(`${module.COMBINER}: KEEPS_PARTIAL_WORK is not declared - a combiner that ` +
        `does not say what it does with work already done is not a declaring unit`);
    }
    combiners[module.COMBINER] = module;
  }

  return { sources, combiners, problems };
}

export function resolveCombiner(name, combiners) {
  const module = combiners[name];
  if (module) return [module, null];
  return [null, `combiner "${name}" has no implementation - fail closed ` +
    `(known: ${Object.keys(combiners).sort().join(", ")})`];
}

// Drive one source. `finished` is observed here, never claimed by the source.
export function run(module, { absent = false } = {}) {
  if (absent) return { source: module.NAME, outcome: "absent", records: [], error: null, finished: null };

  const records = [];
  let error = null;
  try {
    for (const record of module.records()) {
      if (!record || record.from !== module.NAME) {
        throw new Error(`a record without a usable "from" reached the collector ` +
          `(got ${JSON.stringify(record?.from)}, expected ${JSON.stringify(module.NAME)})`);
      }
      records.push(record);
    }
  } catch (raised) {
    error = raised.message;
  }

  const finished = error === null;
  let outcome;
  if (!finished && records.length > 0) outcome = "partial";
  else if (!finished) outcome = "failed";
  else if (records.length > 0) outcome = "worked";
  else outcome = "empty";

  return { source: module.NAME, outcome, records, error, finished };
}

export function runAll(sources, { absent = [] } = {}) {
  return Object.keys(sources).sort()
    .map((name) => run(sources[name], { absent: absent.includes(name) }));
}

// How many of the records the reader ends up holding came from a source that
// did not finish. This is the number a count of records cannot produce.
export function fromUnfinished(runs, kept) {
  const unfinished = new Set(runs.filter((r) => r.outcome === "partial").map((r) => r.source));
  return kept.filter((record) => unfinished.has(record.from)).length;
}

// Combine by refusing the whole batch if any source did not finish.
//
// This is what `Promise.all` does, and archaeology 016 measures its cost:
// the records that already arrived are discarded, and the work that produced
// them ran to completion anyway.
export const COMBINER = "all-or-nothing";
export const KEEPS_PARTIAL_WORK = false;

export function combine(runs) {
  const broken = runs.find((run) => run.outcome === "partial" || run.outcome === "failed");
  if (!broken) return { records: runs.flatMap((run) => run.records), discarded: 0, refused: null };
  return {
    records: [],
    discarded: runs.reduce((n, run) => n + run.records.length, 0),
    refused: `${broken.source} (${broken.outcome}): ${broken.error}`,
  };
}

// Combine by keeping every record and keeping each source's outcome beside it.
//
// This is what `Promise.allSettled` does. It keeps the partial work, which is
// only safe because every record carries `from` — without that, the two
// records from a source that did not finish are anonymous in the pile.
export const COMBINER = "settle-each";
export const KEEPS_PARTIAL_WORK = true;

export function combine(runs) {
  return { records: runs.flatMap((run) => run.records), discarded: 0, refused: null };
}

// A source that hands over real records and then breaks.
//
// The two records it yielded are not hypothetical and not wrong. They are in
// the caller's hands before anything goes wrong, and once they are in the pile
// they are indistinguishable from `short-batch`'s two.
export const NAME = "breaks-midway";
export const CAN_FAIL_WITH = ["connection-reset", "timeout"];

export function* records() {
  yield { from: NAME, id: "b-1" };
  yield { from: NAME, id: "b-2" };
  throw new Error("connection-reset");
}

// A source that yields everything it has and finishes.
//
// It reaches no sibling set and no sibling source. It knows nothing about
// how its records will be combined with anyone else's.
export const NAME = "full-batch";
export const CAN_FAIL_WITH = ["unreadable-page"];

export function* records() {
  yield { from: NAME, id: "f-1" };
  yield { from: NAME, id: "f-2" };
  yield { from: NAME, id: "f-3" };
}

// The control.
//
// It yields TWO records and finishes. `breaks-midway` also yields two records
// and then throws. Without this file, "two records" and "broke after two
// records" would be a single observation, and section 3 of the island test
// could not come out badly.
//
// Same role as the in-memory medium in example 011 and `archive-dump` in
// example 015: a case that produces the same number for the opposite reason.
export const NAME = "short-batch";
export const CAN_FAIL_WITH = ["stale-cursor"];

export function* records() {
  yield { from: NAME, id: "s-1" };
  yield { from: NAME, id: "s-2" };
}

// What a person is shown.
//
// The one thing this report may never do is print a record count on its own,
// because that is the number three different situations share.
const pad = (value, width) => String(value).padEnd(width);

export function runs(rows) {
  const lines = ["  source          outcome   records   finished   error"];
  for (const row of rows) {
    const mark = { worked: "ok", empty: "--", partial: "~~", failed: "!!", absent: "  " }[row.outcome];
    lines.push(`  ${mark} ${pad(row.source, 13)} ${pad(row.outcome, 9)} ${pad(row.records.length, 9)} ` +
      `${pad(row.finished === null ? "-" : row.finished, 10)} ${row.error ?? ""}`.trimEnd());
  }
  return lines.join("\n");
}

export function combined(name, result, unfinished) {
  const lines = [];
  if (result.refused) {
    lines.push(`  ${name}: REFUSED - ${result.refused}`);
    lines.push(`  ${result.discarded} record(s) already in hand were discarded, and the work that ` +
      `produced them ran to completion anyway.`);
    return lines.join("\n");
  }
  lines.push(`  ${name}: ${result.records.length} record(s) kept`);
  if (unfinished > 0) {
    lines.push(`  ${unfinished} of them came from a source that did NOT finish - which is a fact ` +
      `about the records, not about the run.`);
  }
  return lines.join("\n");
}

export function whatACountWouldHaveSaid(rows) {
  const twos = rows.filter((row) => row.records.length === 2);
  const lines = ["  what a count alone would have said:"];
  for (const row of twos) {
    lines.push(`    ${pad(row.source, 15)} 2 records - and this is ${row.outcome}`);
  }
  lines.push(`    ${twos.length} sources, one number, ${new Set(twos.map((r) => r.outcome)).size} outcomes`);
  return lines.join("\n");
}

// The island test.
//
//   node src/island_test.mjs
//
// Section 3 is the one the example exists for: two sources produce two records
// each, and only one of them finished. Section 5 is the finding — removing the
// broken source gives more than keeping it, which is the opposite of what
// example 015 measured.
import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";

import * as collect from "./SMS/collect.mjs";
import * as policy from "./SCL/policy.mjs";
import * as allOrNothing from "./TMS/combiners/all_or_nothing.mjs";
import * as settleEach from "./TMS/combiners/settle_each.mjs";

const here = path.dirname(fileURLToPath(import.meta.url));
const CONTRACT = JSON.parse(fs.readFileSync(path.join(here, "FMS", "contract.json"), "utf8"));
const failures = [];
let ran = 0;
const check = (label, ok, detail = "") => {
  ran += 1;
  process.stdout.write(`  ${ok ? "PASS" : "FAIL"}  ${label}${detail ? ` - ${detail}` : ""}\n`);
  if (!ok) failures.push(label);
};
const say = (line = "") => process.stdout.write(`${line}\n`);

const { sources, combiners, problems } = collect.load();
const rows = collect.runAll(sources);
const row = (name) => rows.find((r) => r.source === name);

say("\n== 1. every source and combiner is an island, and FMS matches the tree");
check("loading raised no problems", problems.length === 0, problems.join("; "));
for (const [unit, declared] of Object.entries(CONTRACT.units)) {
  const dir = path.join(here, ...unit.split("/"));
  const onDisk = fs.readdirSync(dir).filter((n) => n.endsWith(".mjs")).sort();
  check(`${unit}: FMS declares what is on disk`,
    JSON.stringify(onDisk) === JSON.stringify([...declared].sort()),
    `disk ${onDisk.join(", ")} | FMS ${[...declared].sort().join(", ")}`);
  for (const file of onDisk) {
    const body = fs.readFileSync(path.join(dir, file), "utf8");
    const siblings = onDisk.filter((n) => n !== file).map((n) => n.replace(/\.mjs$/, ""));
    const reached = siblings.filter((s) => new RegExp(`from\\s+["'][^"']*${s}`).test(body));
    check(`${unit}/${file} reaches no sibling`, reached.length === 0, reached.join(", "));
    check(`${unit}/${file} reaches no other set`, !/from\s+["'][^"']*\/(SMS|DMS|SCL|FMS)\//.test(body));
  }
}

say("\n== 2. five outcomes, and `partial` is one of them");
check("SMS names five outcomes", collect.OUTCOMES.length === 5, collect.OUTCOMES.join(", "));
check("breaks-midway is partial", row("breaks-midway").outcome === "partial");
check("  not `worked` - it did not finish", row("breaks-midway").finished === false);
check("  not `failed` - it produced real records", row("breaks-midway").records.length === 2);
check("full-batch worked", row("full-batch").outcome === "worked" && row("full-batch").finished === true);
const absentRow = collect.run(sources["breaks-midway"], { absent: true });
check("removed, it is absent and nothing ran", absentRow.outcome === "absent" && absentRow.records.length === 0);

say("\n== 3. the control - the same number for the opposite reason");
const short = row("short-batch");
const broke = row("breaks-midway");
check("short-batch yielded two records", short.records.length === 2);
check("and it DID finish - the control", short.finished === true);
check("breaks-midway yielded two records as well", broke.records.length === 2);
check("and it did NOT finish", broke.finished === false);
check("so a count cannot separate them, and the outcome can - 2 == 2, worked != partial",
  short.records.length === broke.records.length && short.outcome !== broke.outcome);

say("\n== 4. the outcome has to travel WITH the records");
const pile = [...short.records, ...broke.records];
check("in one array the four records are indistinguishable by shape",
  new Set(pile.map((r) => Object.keys(r).sort().join(","))).size === 1);
check("every record names the unit that produced it", pile.every((r) => typeof r.from === "string"));
check("so the pile can still be asked how many came from an unfinished source",
  collect.fromUnfinished(rows, pile) === 2, `${collect.fromUnfinished(rows, pile)} of ${pile.length}`);
const unlabelled = {
  NAME: "drill-unlabelled", CAN_FAIL_WITH: ["x"],
  *records() { yield { id: "u-1" }; },
};
const drill4 = collect.run(unlabelled);
check("DRILL: a record with no origin is refused, not counted",
  drill4.outcome === "failed" && /without a usable "from"/.test(drill4.error ?? ""), drill4.error ?? "");

say("\n== 5. removing it gives more than keeping it");
const removed = allOrNothing.combine(collect.runAll(sources, { absent: ["breaks-midway"] }));
const kept = allOrNothing.combine(rows);
const settled = settleEach.combine(rows);
check("removed (island test): 5 records", removed.records.length === 5);
check("present and failing: 0 records", kept.records.length === 0);
check("and 7 already in hand were discarded", kept.discarded === 7);
check("removing the broken source yields MORE than keeping it",
  removed.records.length > kept.records.length, `${removed.records.length} > ${kept.records.length}`);
check("settle-each keeps all 7, 2 of them from a source that did not finish",
  settled.records.length === 7 && collect.fromUnfinished(rows, settled.records) === 2);
check("the two combiners disagree by 7 records on the same runs",
  settled.records.length - kept.records.length === 7);

say("\n== 6. a unit that does not declare itself is refused");
const mute = { COMBINER: "drill-mute", combine: (runs) => ({ records: [], discarded: 0, refused: null }) };
check("DRILL: a combiner with no KEEPS_PARTIAL_WORK is refused",
  collect.load([], [mute]).problems.some((p) => /drill-mute: KEEPS_PARTIAL_WORK/.test(p)));
const silent = { NAME: "drill-silent", CAN_FAIL_WITH: [], *records() {} };
check("DRILL: a source with an empty CAN_FAIL_WITH is refused (改良點 13)",
  collect.load([silent]).problems.some((p) => /drill-silent: CAN_FAIL_WITH is empty/.test(p)));
check("and the honest units raise nothing", collect.load().problems.length === 0);
check("an unknown combiner fails closed",
  collect.resolveCombiner("no-such-combiner", combiners)[1]?.startsWith("combiner \"no-such-combiner\""));
check("SCL's combiner does resolve", collect.resolveCombiner(policy.combinerName(), combiners)[0] !== null);

say("\n== 7. what this cannot see, asserted so it stays measured");
const swallows = {
  NAME: "drill-swallows", CAN_FAIL_WITH: ["connection-reset"],
  *records() {
    yield { from: "drill-swallows", id: "w-1" };
    yield { from: "drill-swallows", id: "w-2" };
    try { throw new Error("connection-reset"); } catch { return; }
  },
};
const swallowed = collect.run(swallows);
check("a source that catches its own failure reports as `worked`", swallowed.outcome === "worked");
check("and is indistinguishable from short-batch by every field this collector reads",
  swallowed.outcome === short.outcome && swallowed.finished === short.finished
    && swallowed.records.length === short.records.length,
  "the limit named in FMS non_goals - if this ever goes red, the limit changed and the text must too");

say("");
if (failures.length > 0) {
  say(`  ${failures.length} FAILED: ${failures.join(" | ")}`);
  process.exitCode = 1;
} else {
  say(`  ${ran} checks passed - ${rows.length} sources, ${Object.keys(combiners).length} combiners`);
}

// Three sources, one of which hands over real records and then breaks.
//
//   node src/main.mjs             the run under the combiner SCL names
//   node src/main.mjs --compare   removing the broken source, against keeping it
//   node src/main.mjs --strict    exit 1 when a run did not finish and the policy is fatal
import * as report from "./DMS/report.mjs";
import * as policy from "./SCL/policy.mjs";
import * as collect from "./SMS/collect.mjs";

const say = (line = "") => process.stdout.write(`${line}\n`);

function combineWith(name, sources, options = {}) {
  const { combiners } = collect.load();
  const [combiner, problem] = collect.resolveCombiner(name, combiners);
  if (problem) throw new Error(problem);
  const rows = collect.runAll(sources, options);
  const result = combiner.combine(rows);
  return { rows, result, unfinished: collect.fromUnfinished(rows, result.records) };
}

function compare(sources) {
  const removed = combineWith("all-or-nothing", sources, { absent: ["breaks-midway"] });
  const kept = combineWith("all-or-nothing", sources);
  const settled = combineWith("settle-each", sources);

  say("\n                                          kept   what the reader ends up holding");
  say(`  removed (island test)   all-or-nothing  ${String(removed.result.records.length).padEnd(6)} ` +
    `${removed.rows.filter((r) => r.outcome === "worked").length} complete sources`);
  say(`  present and failing     all-or-nothing  ${String(kept.result.records.length).padEnd(6)} ` +
    `nothing - ${kept.result.discarded} records discarded`);
  say(`  present and failing     settle-each     ${String(settled.result.records.length).padEnd(6)} ` +
    `${settled.unfinished} of them from a source that did not finish`);
  say("\n  Removing the broken source gives MORE than keeping it.");
  say("  Example 015 measured the opposite: there, removed and broken were the same number.");
}

function main(argv) {
  const { sources, problems } = collect.load();
  if (problems.length > 0) {
    for (const problem of problems) say(`  REFUSED: ${problem}`);
    return 1;
  }

  if (argv.includes("--compare")) {
    compare(sources);
    return 0;
  }

  const { rows, result, unfinished } = combineWith(policy.combinerName(), sources);
  say(`\n  ${policy.describe()}\n`);
  say(report.runs(rows));
  say("");
  say(report.combined(policy.combinerName(), result, unfinished));
  say("");
  say(report.whatACountWouldHaveSaid(rows));

  const partial = rows.filter((row) => row.outcome === "partial");
  if (argv.includes("--strict") && partial.length > 0 && policy.isFatal()) {
    say(`\n  --strict: ${partial.length} source(s) did not finish and this deployment calls that fatal`);
    return 1;
  }
  return 0;
}

process.exitCode = main(process.argv.slice(2));