How to test a partially applied import

Interrupt one import three ways and count the rows after each. Here node import.mjs r-notx failAt=8 left 7 of 12 rows, the same failure inside a transaction left 0, and a kill by PID left 0 rows, a journal file on disk and exit code -1.

Why check this

Run this once the importer handles a good file, and after every change to its batch size or transaction boundaries.

An import is a program, not a statement. It claims a file, writes rows, updates a counter, writes an audit record and tells a caller it finished. A transaction covers the statements inside it and nothing else.

One failure this prevents: exit code 1, an operator who re-sends the file, and a store holding 19 rows for a 12 row file. Step 12 measures it.

Prerequisites

• Node 22 and node:sqlite, experimental in 22.23.2, so every command carries --no-warnings.
• A scratch directory you can delete, one subdirectory per scenario.
• Windows PowerShell for the kill. kill.ps1 stops the PID it started, never an image name.
• The store runs in journal_mode delete, printed by step 1. WAL mode leaves -wal and -shm.

make.mjs builds one scenario:

import { mkdirSync, rmSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';

const LF = String.fromCharCode(10);
const HEAD = 'sku,name,qty';
const ROWS = [
  'S-311,Hinge,4', 'S-118,Bolt,9', 'S-742,Clamp,2',
  'S-205,Nut,7', 'S-960,Washer,5', 'S-034,Pin,3',
  'S-513,Cap,8', 'S-277,Rivet,6', 'S-881,Screw,1',
  'S-149,Plate,10', 'S-628,Rod,12', 'S-400,Stud,11',
];

const dir = process.argv[2];
rmSync(dir, { recursive: true, force: true });
for (const d of ['inbox', 'work', 'done']) mkdirSync(join(dir, d), { recursive: true });
const text = [HEAD, ...ROWS].join(LF) + LF;
writeFileSync(join(dir, 'inbox', 'orders.csv'), text, 'utf8');

const db = new DatabaseSync(join(dir, 'store.db'));
db.exec('CREATE TABLE product (sku TEXT, name TEXT, qty INTEGER, line INTEGER)');
const v = db.prepare('SELECT sqlite_version() v').get().v;
const j = db.prepare('PRAGMA journal_mode').get().journal_mode;
console.log(dir + '  orders.csv ' + Buffer.byteLength(text, 'utf8') + ' bytes, ' + ROWS.length + ' data rows'
  + '  store.db empty, sqlite ' + v + ', journal_mode ' + j);

import.mjs, the importer under test, with the side effects a real one has:

import { existsSync, readFileSync, renameSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';

const opt = { tx: 'no', failAt: '0', haltAt: '0', key: 'rowid', report: 'store' };
const dir = process.argv[2];
for (const a of process.argv.slice(3)) { const [k, v] = a.split('='); opt[k] = v; }
const P = (...p) => join(dir, ...p);
const block = () => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0);
const LF = String.fromCharCode(10);

// 1. claim the file: inbox to work, before a single row is read
if (existsSync(P('inbox', 'orders.csv'))) renameSync(P('inbox', 'orders.csv'), P('work', 'orders.csv'));
const lines = readFileSync(P('work', 'orders.csv'), 'utf8').split(LF).filter((l) => l.length > 0);
const rows = lines.slice(1).map((l, i) => {
  const [sku, name, qty] = l.split(',');
  return { sku, name, qty: Number(qty), line: i + 2 };
});

// 2. the store, and an audit store that is a second database with its own transactions
const db = new DatabaseSync(P('store.db'));
db.exec('CREATE TABLE IF NOT EXISTS product (sku TEXT, name TEXT, qty INTEGER, line INTEGER)');
if (opt.key === 'sku') db.exec('CREATE UNIQUE INDEX IF NOT EXISTS product_sku ON product (sku)');
const ins = db.prepare(opt.key === 'sku'
  ? 'INSERT INTO product VALUES (?,?,?,?) ON CONFLICT (sku) DO UPDATE SET qty = excluded.qty, line = excluded.line'
  : 'INSERT INTO product VALUES (?,?,?,?)');
const audit = new DatabaseSync(P('audit.db'));
audit.exec('CREATE TABLE IF NOT EXISTS import_event (batch INTEGER, rowsLoaded INTEGER)');
const logBatch = audit.prepare('INSERT INTO import_event VALUES (?,?)');

// 3. the progress record, rewritten after every row, outside every transaction
let loaded = 0;
const cache = new Set();
const progress = (state) => writeFileSync(P('progress.json'), JSON.stringify({ file: 'orders.csv', rowsLoaded: loaded, state }));
progress('running');

let open = false;
try {
  if (opt.tx === 'yes') { db.exec('BEGIN'); open = true; }
  for (let b = 0; b < rows.length; b += 3) {
    // batches are sorted by key before insert, so insert order is not file order
    const batch = rows.slice(b, b + 3).sort((x, y) => (x.sku < y.sku ? -1 : 1));
    for (const r of batch) {
      if (loaded + 1 === Number(opt.failAt)) throw new Error('row ' + (loaded + 1) + ' rejected by the rules');
      // haltAt parks the importer at a known row so the kill lands at the same point every run
      if (loaded + 1 === Number(opt.haltAt)) { writeFileSync(P('halt.flag'), 'at row ' + (loaded + 1)); block(); }
      ins.run(r.sku, r.name, r.qty, r.line);
      cache.add(r.sku);
      loaded += 1;
      progress('running');
    }
    logBatch.run(b / 3 + 1, loaded);
  }
  if (open) { db.exec('COMMIT'); open = false; }
  renameSync(P('work', 'orders.csv'), P('done', 'orders.csv'));
  progress('done');
  console.log('Import complete: ' + loaded + ' rows loaded.');
  process.exit(0);
} catch (e) {
  if (open) { db.exec('ROLLBACK'); open = false; }
  if (opt.report === 'counter') {
    console.log('Import complete: ' + loaded + ' rows loaded.');
    process.exit(0);
  }
  console.log('Import failed: ' + e.message + '. Rows loaded: ' + loaded + ', cached skus: ' + cache.size + '.');
  process.exit(1);
}

state.mjs, the check this page delivers, lists the directory before opening the database:

import { existsSync, readdirSync, readFileSync, statSync } from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';

const dir = process.argv[2];
const wantRows = process.argv.includes('rows=yes');
const P = (...p) => join(dir, ...p);
const LF = String.fromCharCode(10);

// journal files are read BEFORE the database is opened: opening one rolls it back
const journals = readdirSync(dir).filter((f) => f.startsWith('store.db') && f !== 'store.db');
const where = ['inbox', 'work', 'done'].find((d) => existsSync(P(d, 'orders.csv')));
const source = where
  ? readFileSync(P(where, 'orders.csv'), 'utf8').split(LF).filter((l) => l.length > 0).length - 1
  : 0;
const prog = existsSync(P('progress.json')) ? JSON.parse(readFileSync(P('progress.json'), 'utf8')) : null;

const db = new DatabaseSync(P('store.db'));
const stored = db.prepare('SELECT count(*) n FROM product').get().n;
const audit = new DatabaseSync(P('audit.db'));
const ev = audit.prepare('SELECT count(*) b, coalesce(max(rowsLoaded),0) r FROM import_event').get();

const signals = [];
if (where !== 'done') signals.push('the source file sits in ' + where + '/');
if (prog && prog.state !== 'done') signals.push('progress.json state is ' + prog.state);
if (journals.length) signals.push('a journal file is next to store.db');
if (stored !== source) signals.push('the store holds ' + stored + ' of ' + source + ' source rows');

console.log(dir);
console.log('  source      orders.csv in ' + where + '/, ' + source + ' data rows');
console.log('  store       product holds ' + stored + ' rows');
console.log('  progress    ' + (prog ? 'rowsLoaded ' + prog.rowsLoaded + ', state ' + prog.state : 'absent'));
console.log('  audit       ' + ev.b + ' batches logged, high water ' + ev.r + ' rows');
console.log('  journal     ' + (journals.length ? journals.map((f) => f + ' ' + statSync(P(f)).size + ' bytes').join(', ') : 'none') + '  (read before opening store.db)');
console.log('  verdict     ' + (signals.length ? 'MID-IMPORT: ' + signals.join('; ') : 'CLEAN'));
const after = readdirSync(dir).filter((f) => f.startsWith('store.db') && f !== 'store.db');
console.log('  journal     ' + (after.length ? after.join(', ') : 'none') + '  (read after opening store.db)');
if (wantRows) {
  const list = db.prepare('SELECT sku, line FROM product ORDER BY rowid').all();
  console.log('  stored in insert order: ' + list.map((r) => r.sku + ' line ' + r.line).join(', '));
}

journal.mjs fills a transaction large enough to spill pages into the database file:

import { existsSync, mkdirSync, statSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';

const [dir, mode] = process.argv.slice(2);
const P = (f) => join(dir, f);
const size = (f) => (existsSync(P(f)) ? statSync(P(f)).size + ' bytes' : 'absent');

if (mode === 'fill') {
  // one page of cache, so the open transaction spills its pages into store.db
  mkdirSync(dir, { recursive: true });
  const db = new DatabaseSync(P('store.db'));
  db.exec('PRAGMA cache_size = 1');
  db.exec('CREATE TABLE IF NOT EXISTS product (sku TEXT, name TEXT, qty INTEGER, line INTEGER)');
  const ins = db.prepare('INSERT INTO product VALUES (?,?,?,?)');
  db.exec('BEGIN');
  for (let i = 1; i <= 20000; i += 1) ins.run('S-' + i, 'Bulk', i, i);
  writeFileSync(P('halt.flag'), 'after 20000 uncommitted rows');
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0);
}

if (mode === 'read') {
  console.log('  before the first open: store.db ' + size('store.db') + ', journal ' + size('store.db-journal'));
  const rows = new DatabaseSync(P('store.db')).prepare('SELECT count(*) n FROM product').get().n;
  console.log('  after the first open:  store.db ' + size('store.db') + ', journal ' + size('store.db-journal') + ', product holds ' + rows + ' rows');
}

kill.ps1 parks the importer at a known row, then stops it by PID. cmd.exe is the caller, so the exit code survives:

param([string]$dir, [string]$cmd)
if (-not (Test-Path $dir)) { New-Item -ItemType Directory -Path $dir | Out-Null }
if (Test-Path "$dir\halt.flag") { Remove-Item "$dir\halt.flag" }
# cmd.exe is the caller, so the exit code of the importer stays readable after the kill
$caller = Start-Process cmd.exe -ArgumentList '/v:on', '/c', "$cmd & echo EXIT=!ERRORLEVEL!" -PassThru -NoNewWindow -RedirectStandardOutput "$dir\caller.txt"
while (-not (Test-Path "$dir\halt.flag")) { Start-Sleep -Milliseconds 50 }
$child = (Get-CimInstance Win32_Process -Filter "ParentProcessId = $($caller.Id)").ProcessId
Set-Content "$dir\killed-pid.txt" $child     # kill the pid you started, never an image name
Stop-Process -Id $child -Force
$caller.WaitForExit()
"parked " + (Get-Content "$dir\halt.flag") + ", then stopped by pid"
Get-ChildItem $dir -Filter store.db* | ForEach-Object { "  " + $_.Name + " " + $_.Length + " bytes" }
"  the caller read " + (Get-Content "$dir\caller.txt").Trim()

Steps

1. 1

   Step 1.

   Build a directory per scenario, each with the same file and an empty store.

   for d in r-clean r-notx r-tx r-kill r-e1 r-e2 r-e3 r-re-rowid r-re-sku; do node --no-warnings make.mjs $d; done
   

   r-clean  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-notx  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-tx  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-kill  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-e1  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-e2  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-e3  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-re-rowid  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete
   r-re-sku  orders.csv 175 bytes, 12 data rows  store.db empty, sqlite 3.51.3, journal_mode delete

2. 2

   Step 2.

   Import one with nothing going wrong, as the control.

   node --no-warnings import.mjs r-clean
   

   Import complete: 12 rows loaded.

   Twelve rows, the file moved to done/, the progress record closed.

3. 3

   Step 3.

   Reject the eighth row with no transaction open.

   node --no-warnings import.mjs r-notx failAt=8
   

   Import failed: row 8 rejected by the rules. Rows loaded: 7, cached skus: 7.

   Seven rows committed one at a time and stayed.

4. 4

   Step 4.

   Read which rows survived, in insert order.

   node --no-warnings state.mjs r-notx rows=yes | tail -1
   

     stored in insert order: S-118 line 3, S-311 line 2, S-742 line 4, S-034 line 7, S-205 line 5, S-960 line 6, S-277 line 9

   Line 8 is missing and line 9 is there: batches of three are sorted by sku, so the survivors are not the first seven lines.

5. 5

   Step 5.

   Repeat that failure with a transaction open around the loop.

   node --no-warnings import.mjs r-tx failAt=8 tx=yes
   

   Import failed: row 8 rejected by the rules. Rows loaded: 7, cached skus: 7.

   Same message, same counter, empty store. The process cached 7 skus it never stored.

6. 6

   Step 6.

   Kill the importer by PID while its transaction is open.

   powershell -File kill.ps1 r-kill "node --no-warnings import.mjs r-kill tx=yes haltAt=8"
   

   parked at row 8, then stopped by pid
   store.db 8192 bytes
   store.db-journal 4616 bytes
   the caller read EXIT=-1

   The database file is unchanged, a journal sits next to it, and the caller read -1 with no message.

7. 7

   Step 7.

   Kill a second writer whose transaction spills pages into the file.

   powershell -File kill.ps1 r-spill "node --no-warnings journal.mjs r-spill fill"
   

   parked after 20000 uncommitted rows, then stopped by pid
   store.db 495616 bytes
   store.db-journal 9728 bytes
   the caller read EXIT=-1

   This one died with 495616 bytes of database behind it.

8. 8

   Step 8.

   Read the first eight bytes of each journal, then open each database.

   for d in r-kill r-spill; do echo "$d"; xxd -l 8 $d/store.db-journal; node --no-warnings journal.mjs $d read; done
   

   r-kill
   00000000: 0000 0000 0000 0000                      ........
   before the first open: store.db 8192 bytes, journal 4616 bytes
   after the first open:  store.db 8192 bytes, journal 4616 bytes, product holds 0 rows
   r-spill
   00000000: d9d5 05f9 20a1 63d7                      .... .c.
   before the first open: store.db 495616 bytes, journal 9728 bytes
   after the first open:  store.db 8192 bytes, journal absent, product holds 0 rows

   The spilled writer left the SQLite magic, and the first open rolled 495616 bytes back to 8192 and deleted it. The parked one left a zero header the open ignored.

9. 9

   Step 9.

   Read every other place the import wrote.

   for d in r-notx r-tx; do node --no-warnings state.mjs $d; done
   

   r-notx
   source      orders.csv in work/, 12 data rows
   store       product holds 7 rows
   progress    rowsLoaded 7, state running
   audit       2 batches logged, high water 6 rows
   journal     none  (read before opening store.db)
   verdict     MID-IMPORT: the source file sits in work/; progress.json state is running; the store holds 7 of 12 source rows
   journal     none  (read after opening store.db)
   r-tx
   source      orders.csv in work/, 12 data rows
   store       product holds 0 rows
   progress    rowsLoaded 7, state running
   audit       2 batches logged, high water 6 rows
   journal     none  (read before opening store.db)
   verdict     MID-IMPORT: the source file sits in work/; progress.json state is running; the store holds 0 of 12 source rows
   journal     none  (read after opening store.db)

   In r-tx the store holds 0 rows, progress says 7, the audit database says 6, and the file sits claimed in work/.

10. 10

    Step 10.

    Read what the caller was told in three runs.

    for m in "r-e1 failAt=8" "r-e2 failAt=8 tx=yes" "r-e3 failAt=8 tx=yes report=counter"; do d=${m%% *}; node --no-warnings import.mjs $m; echo "  exit $? from $m"; node --no-warnings state.mjs $d | sed -n "3p"; done
    

    Import failed: row 8 rejected by the rules. Rows loaded: 7, cached skus: 7.
    exit 1 from r-e1 failAt=8
    store       product holds 7 rows
    Import failed: row 8 rejected by the rules. Rows loaded: 7, cached skus: 7.
    exit 1 from r-e2 failAt=8 tx=yes
    store       product holds 0 rows
    Import complete: 7 rows loaded.
    exit 0 from r-e3 failAt=8 tx=yes report=counter
    store       product holds 0 rows

    The first exits 1 with 7 rows committed. The third prints success and exits 0 over an empty store, from its counter.

11. 11

    Step 11.

    Run the check on a clean store and on the one the kill left.

    for d in r-clean r-kill; do node --no-warnings state.mjs $d; done
    

    r-clean
    source      orders.csv in done/, 12 data rows
    store       product holds 12 rows
    progress    rowsLoaded 12, state done
    audit       4 batches logged, high water 12 rows
    journal     none  (read before opening store.db)
    verdict     CLEAN
    journal     none  (read after opening store.db)
    r-kill
    source      orders.csv in work/, 12 data rows
    store       product holds 0 rows
    progress    rowsLoaded 7, state running
    audit       2 batches logged, high water 6 rows
    journal     store.db-journal 4616 bytes  (read before opening store.db)
    verdict     MID-IMPORT: the source file sits in work/; progress.json state is running; a journal file is next to store.db; the store holds 0 of 12 source rows
    journal     store.db-journal  (read after opening store.db)

    Four signals separate them: file location, progress state, journal, row count.

12. 12

    Step 12.

    Re-send the file to each store a failure left behind.

    for m in "r-kill none rowid" "r-re-rowid failAt=8 rowid" "r-re-sku failAt=8 sku"; do set -- $m; [ $2 = none ] || node --no-warnings import.mjs $1 $2 key=$3 > /dev/null; node --no-warnings import.mjs $1 key=$3; node --no-warnings state.mjs $1 | sed -n "3p"; done
    

    Import complete: 12 rows loaded.
    store       product holds 12 rows
    Import complete: 12 rows loaded.
    store       product holds 19 rows
    Import complete: 12 rows loaded.
    store       product holds 12 rows

    The killed store re-ran to 12. Over the 7 row store the rowid key ended at 19 rows for a 12 row file, the sku key at 12.

How to read the result

| What you see | What it means | What to do | | --- | --- | --- | | Fewer rows than the file and no transaction in the importer | Every insert committed on its own | Step 3 kept 7 of 12. Key the insert on the data before anyone re-sends the file. | | An empty store and a counter that says otherwise | The rollback covered the table and nothing else | Step 9: progress 7, audit 6, store 0, from one run. | | A store.db-journal file next to the database | A writer died with a transaction open | Step 8. Read the directory before opening the database. | | A journal whose first 8 bytes are zero | It was never made valid for rollback, so the next open ignores it and the row count in front of you is final | Step 8: the file stayed at 8192 bytes and the open left the journal there. | | A journal starting d9d5 05f9 20a1 63d7 | Pages were already written, so the next open rolls them back | Step 8: 495616 bytes went back to 8192 and the journal was deleted. | | Exit code -1 and no message | The process was killed before it could report | Step 6. Treat a silent exit as unfinished, not as done. | | A success line and exit 0 over an empty store | The report reads the counter, not the table | Step 10. Report the stored count. | | The source file in neither the inbox nor the done directory | The importer claimed the file and never released it | Step 9: work/ holds it after both error runs. | | A row for line 9 and none for line 8 | The importer reorders rows inside a batch | Step 4. Read which rows landed rather than assuming the first N. |

Common mistakes

Thresholds

What to check next

• How to test csv import validation: which rows an importer refuses before this starts.
• How to check duplicate rows on import: the rule that decides whether a re-send is safe.
• How to count rows in a csv file: the source count to compare the store against.
• How to check database integrity: what to run on a store left mid-write.
• How to test graceful shutdown: the same interruption against a worker.

FAQ

How to test partial import failure?

Interrupt the importer at one known row three ways: an error with no transaction, the same error inside one, and a kill by PID. Count the rows after each. Here: 7, 0 and 0.

What happens when an import fails halfway?

Whatever the transaction did not cover keeps its last value. Step 9: 0 rows stored, the progress file at 7, the audit database at 6, the source file still claimed.

How to check if an import was rolled back?

Count the stored rows against the source file rather than reading the report. Step 11 also reads the progress record and the journal, because a rolled back store looks like one that never ran.

Csv import partial data: which rows landed?

Not always the first N lines. Step 4 stored line 9 and not line 8: the importer sorts each batch of three before inserting.

Is it safe to re-send a failed import?

Only where the insert is keyed on data. Step 12 re-sent one file to a 7 row store: 19 rows keyed on the rowid, 12 on the sku. Comparing two exports afterwards is How to compare two csv files.

Verified