// Simulador de la recuperación de ARIES sobre un log escrito a mano.
// No es un motor: no hay archivos, páginas de verdad ni buffer pool. Recibe
// un arreglo de registros del log y el estado de las páginas en disco, y
// calcula lo que harían las tres fases (analysis, redo y undo), paso a paso.
//
// Registros (todos con lsn y type):
//   { type: 'update', txn, page, item, before, after, prevLSN }
//   { type: 'clr', txn, page, item, after, prevLSN, undoNextLSN }
//   { type: 'commit' | 'abort' | 'end', txn, prevLSN }
//   { type: 'begin_checkpoint' }
//   { type: 'end_checkpoint', tt: { T1: lastLSN, ... }, dpt: { P5: recLSN, ... } }
// prevLSN y undoNextLSN valen null cuando no hay registro anterior.
// before, after e item son opcionales: si están, el simulador también lleva
// los valores de cada página.
//
// Disco: { P5: { pageLSN: 0, items: { ... } }, ... }

const LOGGED = new Set(['update', 'clr']);

// Índice del begin_checkpoint del último checkpoint completo (con su
// end_checkpoint en el log). Es lo que guardaría el registro maestro.
function lastCheckpoint(log) {
  let begin = -1;
  let found = -1;
  log.forEach((r, i) => {
    if (r.type === 'begin_checkpoint') begin = i;
    if (r.type === 'end_checkpoint' && begin >= 0) found = begin;
  });
  return found;
}

const describe = (r) => {
  if (r.type === 'update') return `${r.txn} update ${r.page}`;
  if (r.type === 'clr') return `${r.txn} CLR ${r.page}`;
  if (r.txn) return `${r.txn} ${r.type}`;
  return r.type;
};

// Fase 1. Lee el log desde el último begin_checkpoint (o desde el principio)
// y reconstruye la transaction table (tt) y la dirty page table (dpt).
export function analysis(log) {
  const from = Math.max(lastCheckpoint(log), 0);
  const tt = new Map(); // txn -> { status, lastLSN }
  const dpt = new Map(); // page -> recLSN
  const ended = new Set();
  const steps = [];

  for (const r of log.slice(from)) {
    const notes = [];
    if (r.type === 'end_checkpoint') {
      // Lo que ya se vio después del begin_checkpoint es más reciente.
      for (const [txn, lastLSN] of Object.entries(r.tt ?? {})) {
        if (!tt.has(txn) && !ended.has(txn)) tt.set(txn, { status: 'running', lastLSN });
      }
      for (const [page, recLSN] of Object.entries(r.dpt ?? {})) {
        dpt.set(page, Math.min(recLSN, dpt.get(page) ?? Infinity));
      }
      notes.push('se cargan la TT y la DPT del checkpoint');
    } else if (r.txn) {
      if (r.type === 'end') {
        tt.delete(r.txn);
        ended.add(r.txn);
        notes.push(`${r.txn} sale de la TT`);
      } else {
        const status = { commit: 'committed', abort: 'aborting' }[r.type]
          ?? tt.get(r.txn)?.status ?? 'running';
        if (!tt.has(r.txn)) notes.push(`${r.txn} entra a la TT`);
        tt.set(r.txn, { status, lastLSN: r.lsn });
        if (r.type === 'commit') notes.push(`${r.txn} queda committed`);
        if (r.type === 'abort') notes.push(`${r.txn} queda aborting`);
        notes.push(`lastLSN(${r.txn}) = ${r.lsn}`);
      }
      if (LOGGED.has(r.type) && !dpt.has(r.page)) {
        dpt.set(r.page, r.lsn);
        notes.push(`${r.page} entra a la DPT con recLSN ${r.lsn}`);
      }
    }
    steps.push({ lsn: r.lsn, record: describe(r), notes });
  }

  const losers = [...tt].filter(([, e]) => e.status !== 'committed').map(([t]) => t);
  const redoLSN = dpt.size ? Math.min(...dpt.values()) : null;
  return { start: log[from]?.lsn ?? null, tt, dpt, losers, redoLSN, steps };
}

function copyDisk(disk) {
  const pages = new Map();
  for (const [page, p] of Object.entries(disk)) {
    pages.set(page, { pageLSN: p.pageLSN, items: { ...p.items } });
  }
  return pages;
}

function pageOf(pages, page) {
  if (!pages.has(page)) pages.set(page, { pageLSN: 0, items: {} });
  return pages.get(page);
}

function apply(page, r) {
  if (r.item !== undefined) page.items[r.item] = r.after;
  page.pageLSN = r.lsn;
}

// Fase 2. Repite la historia desde redoLSN. Un update o un CLR se rehace
// salvo que (a) su página no esté en la DPT, (b) su LSN sea menor que el
// recLSN de la página o (c) el pageLSN ya sea mayor o igual que su LSN.
// (a) y (b) se deciden sin leer la página; (c) obliga a leerla.
export function redo(log, { dpt, redoLSN }, disk) {
  const pages = copyDisk(disk);
  const steps = [];
  if (redoLSN === null) return { pages, steps };
  for (const r of log) {
    if (r.lsn < redoLSN || !LOGGED.has(r.type)) continue;
    let reason = null;
    if (!dpt.has(r.page)) reason = `${r.page} no está en la DPT`;
    else if (r.lsn < dpt.get(r.page)) reason = `${r.lsn} < recLSN ${dpt.get(r.page)}`;
    else {
      const page = pageOf(pages, r.page);
      if (page.pageLSN >= r.lsn) reason = `pageLSN ${page.pageLSN} ≥ ${r.lsn}`;
      else apply(page, r);
    }
    steps.push({ lsn: r.lsn, record: describe(r), action: reason ? 'skip' : 'redo', reason });
  }
  return { pages, steps };
}

// Fase 3. Deshace a las perdedoras, siempre por el mayor LSN pendiente.
// Escribe un CLR por cada update deshecho y un end por cada transacción que
// termina. Devuelve los registros nuevos; no modifica el log que recibe.
export function undo(log, { tt, losers }, pages = new Map(), { step = 10 } = {}) {
  const byLSN = new Map(log.map((r) => [r.lsn, r]));
  let next = Math.max(...log.map((r) => r.lsn)) + step;
  const records = [];
  const steps = [];
  const lastLSN = new Map([...tt].map(([t, e]) => [t, e.lastLSN]));
  const write = (r) => {
    const rec = { lsn: next, ...r };
    next += step;
    records.push(rec);
    lastLSN.set(rec.txn, rec.lsn);
    return rec;
  };

  // Las que hicieron commit y no alcanzaron a escribir su end.
  for (const [txn, e] of tt) {
    if (e.status === 'committed') write({ type: 'end', txn, prevLSN: lastLSN.get(txn) });
  }

  const toUndo = new Set(losers.map((t) => tt.get(t).lastLSN));
  while (toUndo.size) {
    const lsn = Math.max(...toUndo);
    toUndo.delete(lsn);
    const r = byLSN.get(lsn);
    let then;
    if (r.type === 'update') {
      const clr = write({
        type: 'clr', txn: r.txn, page: r.page, item: r.item, after: r.before,
        prevLSN: lastLSN.get(r.txn), undoNextLSN: r.prevLSN,
      });
      apply(pageOf(pages, r.page), clr);
      then = r.prevLSN;
      steps.push({ lsn, record: describe(r), action: 'undo', clr: clr.lsn, undoNextLSN: clr.undoNextLSN });
    } else if (r.type === 'clr') {
      then = r.undoNextLSN;
      steps.push({ lsn, record: describe(r), action: 'jump', undoNextLSN: then });
    } else {
      then = r.prevLSN; // abort: no cambió nada
      steps.push({ lsn, record: describe(r), action: 'skip', undoNextLSN: then });
    }
    if (then === null || then === undefined) {
      const end = write({ type: 'end', txn: r.txn, prevLSN: lastLSN.get(r.txn) });
      steps.at(-1).end = end.lsn;
    } else {
      toUndo.add(then);
    }
  }
  return { records, steps, pages };
}

// Las tres fases seguidas. El log que queda es el de antes más lo que
// escribió el undo; sirve para simular una segunda caída.
export function restart(log, disk, options) {
  const a = analysis(log);
  const r = redo(log, a, disk);
  const u = undo(log, a, r.pages, options);
  return { analysis: a, redo: r, undo: u, log: [...log, ...u.records] };
}

#!/usr/bin/env bash
# Caída y recuperación en un cluster de prueba. Crea un cluster nuevo en
# /tmp (puerto 5415, no el del curso), hace cambios, lo detiene en modo
# immediate (como un corte de luz), lo vuelve a arrancar y muestra lo que
# PostgreSQL escribió en su log durante la recuperación. Al final lo borra.
# El servidor del curso no se toca.
#
#   bash crash.sh        (dentro del contenedor, como root o como postgres;
#                         make crash hace lo mismo)
# Con PORT=5515 bash crash.sh usa otro puerto (las pruebas lo hacen así).
set -eu

PORT="${PORT:-5415}"
DIR="/tmp/w15-crash-$PORT"
LOG="$DIR.log"

# initdb y pg_ctl no aceptan correr como root
if [ "$(id -u)" = 0 ]; then AS="gosu postgres"; else AS=""; fi
sql() { $AS psql -X -q -At -h /tmp -p "$PORT" -d postgres -v ON_ERROR_STOP=1 "$@"; }
cleanup() {
    $AS pg_ctl -D "$DIR" -m immediate stop >/dev/null 2>&1 || true
    rm -rf "$DIR" "$LOG"
}
if pg_isready -q -h /tmp -p "$PORT"; then
    echo "Ya hay un servidor en el puerto $PORT; usen otro: PORT=5499 bash crash.sh" >&2
    exit 1
fi
trap cleanup EXIT

cleanup
$AS initdb -D "$DIR" -U postgres -A trust --no-instructions >/dev/null
$AS pg_ctl -D "$DIR" -l "$LOG" -w -o "-p $PORT -k /tmp -c listen_addresses=''" start >/dev/null

echo "== Antes de la caída"
sql -c "CREATE TABLE payment (payment_id int PRIMARY KEY, amount numeric(8,2) NOT NULL)" \
    -c "CHECKPOINT"
redo=$(sql -c 'SELECT redo_lsn FROM pg_control_checkpoint()')
echo "redo_lsn del último checkpoint: $redo"
sql -c "INSERT INTO payment SELECT i, 10 FROM generate_series(1, 1000) AS i"
echo "commit de 1000 filas; WAL insertado hasta $(sql -c 'SELECT pg_current_wal_insert_lsn()')"

# Una transacción que queda abierta: cambia todas las filas y no hace commit
( printf '%s\n' "BEGIN;" "UPDATE payment SET amount = 0;" "SELECT pg_sleep(60);" \
    | $AS psql -X -q -h /tmp -p "$PORT" -d postgres >/dev/null 2>&1 ) &
for _ in $(seq 1 50); do
    n=$(sql -c "SELECT count(*) FROM pg_stat_activity WHERE query LIKE 'SELECT pg_sleep%'")
    [ "$n" = 1 ] && break
    sleep 0.1
done
xid=$(sql -c "SELECT backend_xid FROM pg_stat_activity WHERE query LIKE 'SELECT pg_sleep%'")
echo "transacción $xid abierta: UPDATE payment SET amount = 0 (sin commit)"
echo "WAL insertado hasta $(sql -c 'SELECT pg_current_wal_insert_lsn()')"

echo "== Caída: pg_ctl -m immediate stop"
$AS pg_ctl -D "$DIR" -m immediate stop >/dev/null
wait || true

echo "== Arranque (líneas nuevas del log del servidor)"
lines=$(wc -l < "$LOG")
$AS pg_ctl -D "$DIR" -l "$LOG" -w -o "-p $PORT -k /tmp -c listen_addresses=''" start >/dev/null
tail -n +"$((lines + 1))" "$LOG" | sed 's/^[0-9-]* [0-9:.]* [A-Z0-9+-]* \[[0-9]*\] //'

echo "== Después de la recuperación"
echo "filas y total: $(sql -F ' ' -c 'SELECT count(*), sum(amount) FROM payment')"
echo "filas con xmax = $xid: $(sql -c "SELECT count(*) FROM payment WHERE xmax = '$xid'")"
echo "estado de la transacción $xid: $(sql -c "SELECT pg_xact_status('$xid')")"
sql -c "CREATE EXTENSION pg_walinspect"
sql -F ' ' -c "SELECT count(*), count(*) FILTER (WHERE record_type IN ('COMMIT', 'ABORT'))
               FROM pg_get_wal_records_info('$redo', pg_current_wal_lsn())
               WHERE xid = '$xid'" | {
    read -r total fin
    echo "registros del WAL con xid $xid: $total; de ellos COMMIT o ABORT: $fin"
}

// El ejemplo resuelto del apunte: el log al momento de la caída y lo que
// había en disco. P1 y P7 son páginas de trip; P3 y P5, de shipment. Cada
// update cambia un ítem de la página (una columna de una fila).
export const LOG = [
  { lsn: 10, type: 'update', txn: 'T1', page: 'P5', item: 'shipment 501.status', before: 'in_transit', after: 'delivered', prevLSN: null },
  { lsn: 20, type: 'update', txn: 'T2', page: 'P3', item: 'shipment 233.price', before: 30, after: 35, prevLSN: null },
  { lsn: 30, type: 'begin_checkpoint' },
  { lsn: 40, type: 'end_checkpoint', tt: { T1: 10, T2: 20 }, dpt: { P5: 10 } },
  { lsn: 50, type: 'update', txn: 'T1', page: 'P3', item: 'shipment 240.price', before: 18.5, after: 20, prevLSN: 10 },
  { lsn: 60, type: 'update', txn: 'T2', page: 'P7', item: 'trip 12.vehicle_id', before: 11, after: 7, prevLSN: 20 },
  { lsn: 70, type: 'commit', txn: 'T2', prevLSN: 60 },
  { lsn: 80, type: 'update', txn: 'T1', page: 'P5', item: 'shipment 502.status', before: 'registered', after: 'in_transit', prevLSN: 50 },
  { lsn: 90, type: 'update', txn: 'T3', page: 'P1', item: 'trip 40.vehicle_id', before: 1, after: 3, prevLSN: null },
  { lsn: 100, type: 'end', txn: 'T2', prevLSN: 70 },
];

// En disco: P3 se escribió después del cambio 50 (steal: T1 no hizo commit);
// las demás páginas nunca se escribieron con estos cambios.
export const DISK = {
  P1: { pageLSN: 0, items: { 'trip 40.vehicle_id': 1 } },
  P3: { pageLSN: 50, items: { 'shipment 233.price': 35, 'shipment 240.price': 20 } },
  P5: { pageLSN: 0, items: { 'shipment 501.status': 'in_transit', 'shipment 502.status': 'registered' } },
  P7: { pageLSN: 0, items: { 'trip 12.vehicle_id': 11 } },
};

// Muestra las tres fases de ARIES sobre el ejemplo del apunte.
//   node trace.mjs           la recuperación después de la caída
//   node trace.mjs 130       además, una segunda caída justo después de que
//                            el undo escribió el registro 130, y la segunda
//                            recuperación
import { restart } from './aries.mjs';
import { LOG, DISK } from './example.mjs';

const dash = (v) => (v === null || v === undefined ? '–' : v);
const table = (m, f) => [...m].map(([k, v]) => `${k}: ${f(v)}`).join(', ');

function show(log, disk) {
  const { analysis: a, redo: r, undo: u } = restart(log, disk);
  console.log(`Analysis (desde el LSN ${a.start})`);
  for (const s of a.steps) {
    console.log(`  ${String(s.lsn).padStart(3)}  ${s.record.padEnd(18)} ${s.notes.join('; ')}`);
  }
  console.log(`  TT  = {${table(a.tt, (e) => `${e.status} ${e.lastLSN}`)}}`);
  console.log(`  DPT = {${table(a.dpt, (v) => v)}}`);
  console.log(`  perdedoras: ${a.losers.join(', ')}; el redo empieza en ${a.redoLSN}`);

  console.log('Redo');
  for (const s of r.steps) {
    const what = s.action === 'redo' ? 'se rehace' : `se salta (${s.reason})`;
    console.log(`  ${String(s.lsn).padStart(3)}  ${s.record.padEnd(18)} ${what}`);
  }

  console.log('Undo');
  for (const s of u.steps) {
    let what;
    if (s.action === 'undo') what = `CLR ${s.clr}, undoNextLSN ${dash(s.undoNextLSN)}`;
    else if (s.action === 'jump') what = `se salta a undoNextLSN ${dash(s.undoNextLSN)}`;
    else what = `se sigue con ${dash(s.undoNextLSN)}`;
    if (s.end) what += `; end en ${s.end}`;
    console.log(`  ${String(s.lsn).padStart(3)}  ${s.record.padEnd(18)} ${what}`);
  }
  console.log('Registros nuevos');
  for (const n of u.records) {
    const extra = n.type === 'clr' ? ` ${n.page}, undoNextLSN ${dash(n.undoNextLSN)}` : '';
    console.log(`  ${n.lsn}  ${n.txn} ${n.type === 'clr' ? 'CLR' : n.type}${extra}  (prevLSN ${dash(n.prevLSN)})`);
  }
  console.log('Páginas al final');
  for (const [page, p] of [...u.pages].sort()) {
    const items = Object.entries(p.items).map(([k, v]) => `${k} = ${v}`).join(', ');
    console.log(`  ${page} (pageLSN ${p.pageLSN}): ${items}`);
  }
  return u.records;
}

const written = show(LOG, DISK);
const cut = Number(process.argv[2]);
if (cut) {
  // Segunda caída: llegaron al log los registros nuevos hasta el LSN cut;
  // las páginas en disco siguen como antes de la primera recuperación.
  console.log(`\n---- segunda caída después del registro ${cut} ----\n`);
  show([...LOG, ...written.filter((n) => n.lsn <= cut)], DISK);
}

-- Semana 15: el WAL de PostgreSQL con pg_walinspect.
-- Se corre con psql sobre chasqui_w15 (make la crea con las extensiones):
--   psql -d chasqui_w15 -f wal.sql
-- Usa \gset para guardar LSNs en variables de psql. Como el servidor del
-- contenedor puede tener otras bases con actividad, de cada rango se muestran
-- solo los registros de esta base (o de la transacción). Cambia el envío 501 y
-- agrega un evento: para repetirlo desde cero, make otra vez.

\echo '== Configuración del WAL'
SELECT name, setting, unit
FROM pg_settings
WHERE name IN ('wal_level', 'synchronous_commit', 'full_page_writes',
               'data_checksums', 'checkpoint_timeout', 'max_wal_size',
               'wal_segment_size')
ORDER BY name;

SELECT '%/' || oid || '/%' AS esta_base FROM pg_database
WHERE datname = current_database() \gset

\echo '== Dónde estamos en el WAL'
SELECT pg_current_wal_lsn() AS lsn,
  pg_walfile_name(pg_current_wal_lsn()) AS segment;

\echo '== Un cambio justo después de un checkpoint'
CHECKPOINT;
SELECT redo_lsn, checkpoint_lsn FROM pg_control_checkpoint();
SELECT pg_current_wal_insert_lsn() AS antes \gset
BEGIN;
INSERT INTO shipment_event (shipment_id, seq, at, agency_id, kind)
SELECT shipment_id, seq + 1, now(), agency_id, kind
FROM shipment_event
WHERE shipment_id = 501
ORDER BY seq DESC
LIMIT 1;
UPDATE shipment SET price = price + 1 WHERE shipment_id = 501;
SELECT pg_current_xact_id() AS xid \gset
COMMIT;
SELECT pg_current_wal_insert_lsn() AS despues \gset

-- La relación de cada registro sale de su block_ref: rel tablespace/base/relfilenode
SELECT start_lsn, end_lsn, prev_lsn, xid, resource_manager AS rmgr,
  record_type, record_length AS len, fpi_length AS fpi,
  pg_filenode_relation(0, substring(block_ref FROM 'rel \d+/\d+/(\d+)')::oid) AS rel,
  substring(block_ref FROM 'blk (\d+)')::int AS blk
FROM pg_get_wal_records_info(:'antes', :'despues')
WHERE xid = :'xid' OR block_ref LIKE :'esta_base';

\echo '== El mismo UPDATE otra vez, sin checkpoint antes'
SELECT pg_current_wal_insert_lsn() AS antes \gset
UPDATE shipment SET price = price + 1 WHERE shipment_id = 501
RETURNING xmin AS xid \gset
SELECT pg_current_wal_insert_lsn() AS despues \gset

SELECT start_lsn, end_lsn, xid, resource_manager AS rmgr,
  record_type, record_length AS len, fpi_length AS fpi,
  pg_filenode_relation(0, substring(block_ref FROM 'rel \d+/\d+/(\d+)')::oid) AS rel,
  substring(block_ref FROM 'blk (\d+)')::int AS blk
FROM pg_get_wal_records_info(:'antes', :'despues')
WHERE xid = :'xid' OR block_ref LIKE :'esta_base';

-- pg_get_wal_stats no se puede filtrar: cuenta todo el rango, de todas las bases
SELECT "resource_manager/record_type" AS rmgr, count, record_size, fpi_size
FROM pg_get_wal_stats(:'antes', :'despues')
WHERE count > 0;

\echo '== El pageLSN de la página del envío 501'
SELECT ctid FROM shipment WHERE shipment_id = 501;
SELECT lsn AS page_lsn
FROM page_header(get_raw_page('shipment',
  (SELECT (ctid::text::point)[0]::int FROM shipment WHERE shipment_id = 501)));

// Pruebas del simulador de ARIES con el ejemplo resuelto del apunte.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { analysis, redo, undo, restart } from './aries.mjs';
import { LOG, DISK } from './example.mjs';

const plain = (m) => Object.fromEntries(m);

test('analysis empieza en el begin_checkpoint y reconstruye la TT y la DPT', () => {
  const a = analysis(LOG);
  assert.equal(a.start, 30);
  assert.deepEqual(plain(a.tt), {
    T1: { status: 'running', lastLSN: 80 },
    T3: { status: 'running', lastLSN: 90 },
  });
  assert.deepEqual(plain(a.dpt), { P5: 10, P3: 50, P7: 60, P1: 90 });
  assert.deepEqual(a.losers, ['T1', 'T3']);
  assert.equal(a.redoLSN, 10);
});

test('redo: 20 se salta por recLSN, 50 por pageLSN; los demás updates se rehacen', () => {
  const { steps } = redo(LOG, analysis(LOG), DISK);
  assert.deepEqual(steps.map((s) => [s.lsn, s.action]), [
    [10, 'redo'], [20, 'skip'], [50, 'skip'], [60, 'redo'], [80, 'redo'], [90, 'redo'],
  ]);
  assert.match(steps[1].reason, /recLSN 50/);
  assert.match(steps[2].reason, /pageLSN 50/);
});

test('después del redo, las páginas están como en el momento de la caída', () => {
  const { pages } = redo(LOG, analysis(LOG), DISK);
  assert.deepEqual(pages.get('P5'), {
    pageLSN: 80,
    items: { 'shipment 501.status': 'delivered', 'shipment 502.status': 'in_transit' },
  });
  assert.deepEqual(pages.get('P1').items, { 'trip 40.vehicle_id': 3 });
  assert.equal(pages.get('P3').pageLSN, 50);
});

test('redo es idempotente: rehacer sobre lo ya rehecho no cambia nada', () => {
  const a = analysis(LOG);
  const once = redo(LOG, a, DISK).pages;
  const twice = redo(LOG, a, Object.fromEntries(once));
  assert.deepEqual(twice.pages, once);
  assert.ok(twice.steps.every((s) => s.action === 'skip'));
});

test('undo escribe los CLR y los end del apunte', () => {
  const a = analysis(LOG);
  const { records } = undo(LOG, a, redo(LOG, a, DISK).pages);
  assert.deepEqual(
    records.map(({ lsn, type, txn, page, prevLSN, undoNextLSN }) =>
      [lsn, type, txn, page ?? null, prevLSN, undoNextLSN ?? null]),
    [
      [110, 'clr', 'T3', 'P1', 90, null],
      [120, 'end', 'T3', null, 110, null],
      [130, 'clr', 'T1', 'P5', 80, 50],
      [140, 'clr', 'T1', 'P3', 130, 10],
      [150, 'clr', 'T1', 'P5', 140, null],
      [160, 'end', 'T1', null, 150, null],
    ],
  );
});

test('después de la recuperación solo quedan los cambios de T2, que hizo commit', () => {
  const { undo: u, log } = restart(LOG, DISK);
  const items = Object.assign({}, ...[...u.pages.values()].map((p) => p.items));
  assert.deepEqual(items, {
    'trip 40.vehicle_id': 1,
    'shipment 233.price': 35,
    'shipment 240.price': 18.5,
    'shipment 501.status': 'in_transit',
    'shipment 502.status': 'registered',
    'trip 12.vehicle_id': 7,
  });
  assert.equal(log.length, LOG.length + 6);
  assert.equal(LOG.length, 10); // restart no cambia el log que recibe
});

test('segunda caída después del CLR 130: el cambio 80 no se deshace dos veces', () => {
  const first = restart(LOG, DISK).undo.records;
  const log2 = [...LOG, ...first.filter((r) => r.lsn <= 130)];
  const second = restart(log2, DISK);
  assert.deepEqual(plain(second.analysis.tt), { T1: { status: 'running', lastLSN: 130 } });
  assert.deepEqual(second.redo.steps.filter((s) => s.action === 'redo').map((s) => s.lsn),
    [10, 60, 80, 90, 110, 130]);
  assert.deepEqual(second.undo.steps.map((s) => [s.lsn, s.action]),
    [[130, 'jump'], [50, 'undo'], [10, 'undo']]);
  assert.deepEqual(second.undo.records.map((r) => r.lsn), [140, 150, 160]);
  assert.deepEqual(second.undo.pages, restart(LOG, DISK).undo.pages);
});

test('sin checkpoint, analysis empieza al principio del log', () => {
  const log = LOG.filter((r) => !r.type.endsWith('checkpoint'));
  const a = analysis(log);
  assert.equal(a.start, 10);
  assert.deepEqual(plain(a.dpt), { P5: 10, P3: 20, P7: 60, P1: 90 });
});

test('una transacción con commit y sin end no es perdedora; el undo le escribe su end', () => {
  const log = LOG.slice(0, 9); // sin el registro 100 (T2 end)
  const r = restart(log, DISK);
  assert.deepEqual(r.analysis.losers, ['T1', 'T3']);
  assert.equal(r.analysis.tt.get('T2').status, 'committed');
  assert.deepEqual(r.undo.records[0], { lsn: 100, type: 'end', txn: 'T2', prevLSN: 70 });
});

test('una transacción que estaba haciendo rollback sigue desde su abort', () => {
  // T3 hizo abort en 110 y no alcanzó a deshacer nada
  const log = [...LOG, { lsn: 110, type: 'abort', txn: 'T3', prevLSN: 90 }];
  const r = restart(log, DISK);
  assert.equal(r.analysis.tt.get('T3').status, 'aborting');
  assert.deepEqual(r.undo.steps.slice(0, 2).map((s) => [s.lsn, s.action]),
    [[110, 'skip'], [90, 'undo']]);
});

// Corre crash.sh (en otro puerto que make crash) y revisa lo que el apunte
// dice de la caída: lo que hizo commit sobrevive, lo que no, no aparece, y
// el log del servidor muestra la recuperación. No se comparan los LSN ni el
// número de filas que alcanzó a rehacer el UPDATE sin commit: cambian.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { execFileSync } from 'node:child_process';
import { existsSync } from 'node:fs';

const PORT = '5515';
const out = execFileSync('bash', [new URL('./crash.sh', import.meta.url).pathname], {
  env: { ...process.env, PORT }, encoding: 'utf8',
});

test('el log del servidor muestra la recuperación automática', () => {
  assert.match(out, /database system was interrupted/);
  assert.match(out, /database system was not properly shut down; automatic recovery in progress/);
  assert.match(out, /redo done at [0-9A-F]+\/[0-9A-F]+/);
  assert.match(out, /checkpoint starting: end-of-recovery/);
  assert.match(out, /database system is ready to accept connections/);
});

test('el redo empieza en el redo_lsn del último checkpoint', () => {
  const redo = out.match(/redo_lsn del último checkpoint: (\S+)/)[1];
  assert.equal(out.match(/redo starts at (\S+)/)[1], redo);
});

test('sobreviven las 1000 filas con commit; el UPDATE sin commit no aparece', () => {
  assert.match(out, /filas y total: 1000 10000\.00/);
  const xid = out.match(/transacción (\d+) abierta/)[1];
  assert.match(out, new RegExp(`estado de la transacción ${xid}: aborted`));
  assert.match(out, /de ellos COMMIT o ABORT: 0/);
});

test('el cluster de prueba se borró', () => {
  assert.equal(existsSync(`/tmp/w15-crash-${PORT}`), false);
});

// Pruebas de lo que el apunte dice del WAL de PostgreSQL, con las mismas
// sentencias de wal.sql (allí con \gset; aquí con parámetros). El servidor
// del contenedor es compartido: de cada rango del WAL se miran solo los
// registros de esta transacción o de esta base.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';

let client;
let dbPattern;
before(async () => {
  client = new pg.Client();
  await client.connect();
  const { rows } = await client.query(
    "SELECT '%/' || oid || '/%' AS p FROM pg_database WHERE datname = current_database()");
  dbPattern = rows[0].p;
});
after(() => client.end());

const insertLSN = async () =>
  (await client.query('SELECT pg_current_wal_insert_lsn() AS lsn')).rows[0].lsn;

const records = async (from, to, xid) => (await client.query(
  `SELECT start_lsn, end_lsn, prev_lsn, xid::text, resource_manager AS rmgr,
     record_type, fpi_length AS fpi,
     pg_filenode_relation(0, substring(block_ref FROM 'rel \\d+/\\d+/(\\d+)')::oid)::text AS rel,
     substring(block_ref FROM 'blk (\\d+)')::int AS blk
   FROM pg_get_wal_records_info($1, $2)
   WHERE xid = $3::xid OR block_ref LIKE $4`,
  [from, to, xid, dbPattern])).rows;

test('configuración: WAL de 16 MB, full_page_writes, commit síncrono y checksums', async () => {
  const { rows } = await client.query(
    `SELECT name, setting FROM pg_settings
     WHERE name IN ('wal_level', 'synchronous_commit', 'full_page_writes',
                    'data_checksums', 'checkpoint_timeout', 'wal_segment_size')`);
  assert.deepEqual(Object.fromEntries(rows.map((r) => [r.name, r.setting])), {
    checkpoint_timeout: '300', data_checksums: 'on', full_page_writes: 'on',
    synchronous_commit: 'on', wal_level: 'replica', wal_segment_size: '16777216',
  });
});

test('un checkpoint deja CHECKPOINT_REDO en redo_lsn y CHECKPOINT_ONLINE en checkpoint_lsn', async () => {
  await client.query('CHECKPOINT');
  const { rows: [c] } = await client.query('SELECT redo_lsn, checkpoint_lsn FROM pg_control_checkpoint()');
  const type = async (lsn) => (await client.query(
    'SELECT record_type FROM pg_get_wal_record_info($1)', [lsn])).rows[0].record_type;
  assert.equal(await type(c.redo_lsn), 'CHECKPOINT_REDO');
  assert.equal(await type(c.checkpoint_lsn), 'CHECKPOINT_ONLINE');
});

let pageLSNCheck;
test('después de un checkpoint, la primera modificación de cada página lleva la página entera', async () => {
  await client.query('CHECKPOINT');
  const from = await insertLSN();
  await client.query('BEGIN');
  await client.query(
    `INSERT INTO shipment_event (shipment_id, seq, at, agency_id, kind)
     SELECT shipment_id, seq + 1, now(), agency_id, kind
     FROM shipment_event WHERE shipment_id = 501 ORDER BY seq DESC LIMIT 1`);
  await client.query('UPDATE shipment SET price = price + 1 WHERE shipment_id = 501');
  const { rows: [{ xid }] } = await client.query('SELECT pg_current_xact_id()::text AS xid');
  await client.query('COMMIT');
  const to = await insertLSN();
  const mine = (await records(from, to, xid)).filter((r) => r.xid === xid);
  const kinds = mine.map((r) => `${r.rmgr} ${r.record_type} ${r.rel ?? ''}`.trim());
  for (const k of ['Heap INSERT shipment_event', 'Btree INSERT_LEAF shipment_event_pkey',
    'Heap LOCK shipment', 'Heap LOCK agency', 'Transaction COMMIT']) {
    assert.ok(kinds.includes(k), `falta ${k}: ${kinds}`);
  }
  assert.ok(kinds.some((k) => /^Heap (HOT_)?UPDATE shipment$/.test(k)), kinds);
  assert.equal(kinds.at(-1), 'Transaction COMMIT');
  const insert = mine.find((r) => r.record_type === 'INSERT');
  assert.ok(insert.fpi > 0);
  assert.equal(mine.find((r) => r.record_type === 'COMMIT').fpi, 0);
});

const redoLSN = async () =>
  (await client.query('SELECT redo_lsn FROM pg_control_checkpoint()')).rows[0].redo_lsn;

test('el mismo UPDATE sin checkpoint antes: sin full-page write, y pageLSN = end_lsn', async () => {
  // Si otro proceso hace un checkpoint justo antes del UPDATE, el registro
  // lleva la página entera; en ese caso se repite el UPDATE una vez.
  let from, to, xid;
  for (let attempt = 0; attempt < 2; attempt++) {
    const redo = await redoLSN();
    from = await insertLSN();
    ({ rows: [{ xid }] } = await client.query(
      'UPDATE shipment SET price = price + 1 WHERE shipment_id = 501 RETURNING xmin::text AS xid'));
    to = await insertLSN();
    if ((await redoLSN()) === redo) break;
  }
  const rs = await records(from, to, xid);
  const upd = rs.find((r) => r.xid === xid && /UPDATE/.test(r.record_type));
  assert.equal(upd.fpi, 0);
  assert.equal(upd.rel, 'shipment');
  const { rows: [{ blk }] } = await client.query(
    'SELECT (ctid::text::point)[0]::int AS blk FROM shipment WHERE shipment_id = 501');
  assert.equal(upd.blk, blk);
  const { rows: [{ lsn }] } = await client.query(
    "SELECT lsn FROM page_header(get_raw_page('shipment', $1))", [blk]);
  const last = rs.filter((r) => r.rel === 'shipment' && r.blk === blk).at(-1);
  assert.equal(lsn, last.end_lsn);
});

test('prev_lsn apunta al registro anterior del WAL, no al anterior de la transacción', async () => {
  const from = await insertLSN();
  await client.query('UPDATE shipment SET price = price + 1 WHERE shipment_id = 502');
  const to = await insertLSN();
  const { rows } = await client.query(
    'SELECT start_lsn, prev_lsn FROM pg_get_wal_records_info($1, $2)', [from, to]);
  for (let i = 1; i < rows.length; i++) assert.equal(rows[i].prev_lsn, rows[i - 1].start_lsn);
});

# Semana 15: recuperación, el WAL de PostgreSQL y un simulador de ARIES.
# Se usa dentro del contenedor del curso.
#
#   make          crea la base de datos chasqui_w15 con los datos de ejemplo
#                 e instala pg_walinspect y pageinspect
#   make test     la vuelve a crear y corre las pruebas (*.test.mjs); una de
#                 ellas corre crash.sh en un cluster de prueba
#   make crash    caída y recuperación en un cluster de prueba en /tmp, en el
#                 puerto 5415 (no el del curso); al final lo borra
#   make trace    las tres fases de ARIES sobre el ejemplo del apunte
#   make clean    borra chasqui_w15 y lo que haya quedado de crash.sh
#
# wal.sql se corre sobre chasqui_w15 con
#   psql -d chasqui_w15 -f wal.sql

DB = chasqui_w15
PSQL = psql -X -q -v ON_ERROR_STOP=1 -d $(DB)

.PHONY: all test crash trace clean

all:
	chasqui-load $(DB)
	$(PSQL) -c "CREATE EXTENSION IF NOT EXISTS pg_walinspect" \
	        -c "CREATE EXTENSION IF NOT EXISTS pageinspect"

# wal.test.mjs hace CHECKPOINT y crash.test.mjs arranca un cluster: una
# prueba después de otra.
test: all
	PGDATABASE=$(DB) node --test --test-concurrency=1 --test-reporter=spec

crash:
	bash crash.sh

trace:
	node trace.mjs 130

clean:
	dropdb --if-exists $(DB)
	rm -rf /tmp/w15-crash-*