From a73c0b482c5795c71c1762c454b3538f3b753681 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adam=20Zieli=C5=84ski?= Date: Fri, 14 Aug 2026 03:19:36 +0200 Subject: [PATCH] [PHP] Resume file-backed db-pull from flushed cursor boundaries --- CLAUDE.md | 2 +- README.md | 22 +- packages/reprint-client/src/import.php | 324 ++++++--- .../src/lib/pull/class-pull.php | 26 +- tests/Import/DatabaseCommandRestartTest.php | 618 ++++++++++++++++++ tests/e2e/site-registry.json | 6 + ...port-54-db-pull-index-interruption.test.js | 177 +++++ ...rt-59-file-db-pull-artifact-guards.test.js | 300 +++++++++ 8 files changed, 1379 insertions(+), 96 deletions(-) create mode 100644 tests/Import/DatabaseCommandRestartTest.php create mode 100644 tests/e2e/tests/import-54-db-pull-index-interruption.test.js create mode 100644 tests/e2e/tests/import-59-file-db-pull-artifact-guards.test.js diff --git a/CLAUDE.md b/CLAUDE.md index 37e9083f..218ef934 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -168,7 +168,7 @@ Entries in `paths_to_remove` under `wp-content/plugins/` also trigger automatic ### SQL Streaming Source Retries -When one SQL response ends in the middle of a statement, the importer keeps those bytes in memory and requests the next response in the same process. If that process stops while writing to stdout or MySQL, Reprint cannot tell how much SQL the destination received or kept. A new process refuses to continue until the destination is reset or restored and `db-pull --abort` is run. +When one SQL response ends in the middle of a statement, the importer keeps those bytes in memory and requests the next response in the same process. File output can continue in a new process from the last file size and source position Reprint saved. If a process stops while writing to stdout or MySQL, Reprint cannot tell how much SQL the destination received or kept. A new process refuses to continue until the destination is reset or restored and `db-pull --abort` is run. ### Progress Tracking diff --git a/README.md b/README.md index f3630b1e..f03eab84 100644 --- a/README.md +++ b/README.md @@ -367,11 +367,12 @@ The three modes: | `stdout` | Streams SQL to stdout, progress/status goes to stderr | none | | `mysql` | Connects via `mysqli::multi_query()` and executes statements as they arrive | none | -All three modes retry a source crash inside the same importer process. If the -importer itself stops while writing to `stdout` or `mysql`, Reprint cannot tell -how much SQL the receiving program or database got. It refuses to continue. -Reset or restore that destination, abort the pull, download `db.sql`, and use -`db-apply`. +All three modes retry a source crash inside the same importer process. A new +process can continue `file` output from the last file size and source position +Reprint saved. It first removes any later bytes from `db.sql`. If the importer +itself stops while writing to `stdout` or `mysql`, Reprint cannot tell how much +SQL the receiving program or database got. It refuses to continue. Reset or +restore that destination, abort the pull, download `db.sql`, and use `db-apply`. The `mysql` mode requires `--mysql-database` and accepts `--mysql-host`, `--mysql-port`, `--mysql-user`, and `--mysql-password` (or the `MYSQL_PASSWORD` @@ -383,7 +384,7 @@ The command returns one of three exit codes: - 0: sync completed - 1: failure -- 2: partial completion, needs re-running +- 2: partial completion; re-run the same command when its output is resumable #### Step 4 — Download files delta. @@ -669,9 +670,10 @@ are absent while the plan is still being built. #### `/pull/state.json` — the pull state store This is the pull state store. Pull commands read it on startup and write it -back periodically and on shutdown. It stores everything needed to resume after -a crash or interruption: the current command, cursor position, AIMD tuning -state, and per-phase bookmarks. +back periodically and on shutdown. It stores command, cursor, AIMD tuning, and +phase state. Some commands also need the file they were writing or the last +position the other server confirmed. This state file alone cannot continue +direct SQL output. Written atomically (temp file + rename) so a crash mid-write never corrupts it. If the JSON is invalid on load, the importer renames it to @@ -810,4 +812,4 @@ php reprint.phar --state-dir=DIR --fs-root=DIR [options] * `flat-docroot` — Reassemble pulled files into a standard WordPress directory layout using symlinks. Useful when the source site has a non-standard layout (e.g. WP Cloud with ABSPATH separate from wp-content). * `apply-runtime` — Generates server configuration files (`runtime.php`, `start.sh` or `nginx.conf`) from the pull state selected by the remote Reprint API URL. No network calls are made. See [Step 6](#step-6--generate-runtime-configuration). -All commands except `preflight-assert` support `--abort` to abort the current sync and exit. For `files-pull`, this clears sync progress but keeps the remote index and downloaded files — the next run performs a delta sync. For `db-pull` and `db-index`, it clears the output file so the next run starts from scratch. Interrupted commands automatically resume from the last saved cursor. +All commands except `preflight-assert` support `--abort` to abort the current sync and exit. For `files-pull`, this clears sync progress but keeps the remote index and downloaded files — the next run performs a delta sync. For `db-pull` and `db-index`, it clears the output file so the next run starts from scratch. File downloads resume from the last saved cursor. After direct `stdout` or MySQL output stops, reset or restore its external destination before aborting that pull. diff --git a/packages/reprint-client/src/import.php b/packages/reprint-client/src/import.php index 1f4ac942..bb38bab8 100755 --- a/packages/reprint-client/src/import.php +++ b/packages/reprint-client/src/import.php @@ -173,6 +173,8 @@ class ImportClient private const SAVE_STATE_EVERY_N_CHUNKS = 50; private const STATE_PATH_ENCODING_PREFIX = "base64:"; private const SQLITE_PREPARED_INSERT_CACHE_MAX = 128; + private const DATABASE_DUMP_INTENT_FILE = 'database-dump.intent'; + private const DATABASE_DUMP_RECORD_FILE = 'database-dump.json'; private const DATABASE_PULL_OUTPUT_STATUS_FILE = 'database-pull-output.json'; /** @@ -922,10 +924,38 @@ public function run( is_file($output_status_file) && ( $output_status['status'] ?? null ) !== 'complete'; $saved_command = $this->get_state()->active_resumable_command; - $has_unfinished_sql_stage = + $has_unfinished_db_pull = $saved_command->command_name === 'db-pull' - && $saved_command->current_stage === 'sql' && in_array($saved_command->completion_state, ['in_progress', 'partial'], true); + $has_unfinished_sql_stage = + $has_unfinished_db_pull + && $saved_command->current_stage === 'sql'; + if ( + !$abort + && in_array($command, ['db-pull', 'pull', 'pull-db', 'db-apply'], true) + && $has_unfinished_db_pull + && !in_array($this->get_state()->sql_output, ['file', 'stdout', 'mysql'], true) + ) { + throw new RuntimeException( + 'This db-pull did not finish, and Reprint did not save where it was writing ' . + 'SQL. Run db-pull --abort, then start db-pull again.', + ); + } + if ( + !$abort + && in_array($command, ['db-pull', 'pull', 'pull-db', 'db-apply'], true) + && $has_unfinished_db_pull + && $this->get_state()->sql_output === 'file' + && !is_file(wp_join_unix_paths( + $this->pull_state_directory, + self::DATABASE_DUMP_INTENT_FILE, + )) + ) { + throw new RuntimeException( + 'This db-pull did not finish, and Reprint cannot tell whether db.sql belongs ' . + 'to that pull. Run db-pull --abort, then start db-pull again.', + ); + } $saved_output = $this->get_state()->sql_output; if ( !$abort @@ -1071,6 +1101,19 @@ public function run( "Invalid --sql-output mode: {$mode}. Valid modes: file, stdout, mysql", ); } + $saved_sql_output = $this->get_state()->sql_output; + if ( + !$abort + && $has_unfinished_db_pull + && $saved_sql_output !== null + && $saved_sql_output !== $mode + ) { + throw new RuntimeException( + "Cannot change --sql-output from {$saved_sql_output}" . + " to {$mode} while db-pull is unfinished. Resume with the original mode " . + 'or run db-pull --abort.', + ); + } $this->sql_output_mode = $mode; $this->get_state()->sql_output = $mode; } elseif (isset($this->get_state()->sql_output)) { @@ -2283,29 +2326,23 @@ private function handle_abort(string $command): void "RESTART | Clearing db-pull state", true, ); - if ($this->sql_output_mode === "file") { - $sql_file = wp_join_unix_paths($this->state_dir, "db.sql"); - if (file_exists($sql_file)) { - unlink($sql_file); - $this->audit_log( - "FILE DELETE | {$sql_file} | abort db-pull", - ); + foreach ([ + wp_join_unix_paths($this->state_dir, "db.sql"), + wp_join_unix_paths($this->state_dir, "db-tables.jsonl"), + wp_join_unix_paths($this->pull_state_directory, "domains.json"), + wp_join_unix_paths($this->pull_state_directory, "sql-stats.json"), + ] as $path) { + if (file_exists($path)) { + if (!unlink($path)) { + // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- The path belongs to the caller-selected state directory. + throw new RuntimeException( + "Reprint could not delete db-pull file: {$path}. " . + "Check its permissions, then run db-pull --abort again.", + ); + } + $this->audit_log("FILE DELETE | {$path} | abort db-pull"); } } - $tables_file = wp_join_unix_paths($this->state_dir, "db-tables.jsonl"); - if (file_exists($tables_file)) { - unlink($tables_file); - $this->audit_log( - "FILE DELETE | {$tables_file} | abort db-pull", - ); - } - $domains_file = wp_join_unix_paths($this->pull_state_directory, "domains.json"); - if (file_exists($domains_file)) { - unlink($domains_file); - $this->audit_log( - "FILE DELETE | {$domains_file} | abort db-pull", - ); - } $this->clear_database_pull_records(); $this->reset_state(); $this->save_state(); @@ -3799,10 +3836,18 @@ public function run_db_sync(): void { $state_command = $this->get_state()->active_resumable_command->command_name ?? null; $sql_file = wp_join_unix_paths($this->state_dir, "db.sql"); + $dump_intent_file = wp_join_unix_paths( + $this->pull_state_directory, + self::DATABASE_DUMP_INTENT_FILE, + ); $has_progress = $state_command === "db-pull" && - ($this->get_state()->active_resumable_command->completion_state ?? null) === "in_progress"; + in_array( + $this->get_state()->active_resumable_command->completion_state ?? null, + ["in_progress", "partial"], + true, + ); $current_status = $state_command === "db-pull" ? $this->get_state()->active_resumable_command->completion_state ?? null @@ -3829,6 +3874,8 @@ public function run_db_sync(): void } if ($has_progress) { + $this->get_state()->active_resumable_command->completion_state = "in_progress"; + $this->save_state(); $stage = $this->get_state()->active_resumable_command->current_stage ?? "db-index"; $this->audit_log( sprintf( @@ -3857,7 +3904,25 @@ public function run_db_sync(): void $this->get_state()->active_resumable_command->current_stage = "db-index"; $this->get_state()->diff = new FileDiffProgressState(); $this->get_state()->db_index = new DatabaseTableIndexState(); + $this->get_state()->sql_bytes = null; $this->get_state()->sql_output = $this->sql_output_mode; + if ($this->sql_output_mode === 'file') { + // Remove the old db.sql before removing the information which marks it + // complete. If the process stops here, either both still exist or db.sql is gone. + if (file_exists($sql_file)) { + if (!unlink($sql_file)) { + throw new RuntimeException( + "Reprint could not remove the old db.sql: {$sql_file}. " . + "Check its permissions, then start db-pull again.", + ); + } + $this->audit_log("FILE DELETE | {$sql_file} | start db-pull"); + } + $this->clear_database_pull_records(); + $this->write_json_file($dump_intent_file, [ + 'create_table_query' => true, + ]); + } $this->save_state(); $this->audit_log("START db-pull", true); @@ -3927,10 +3992,40 @@ public function run_db_sync(): void return; } - // Mark as complete + if ($this->sql_output_mode === 'file') { + $dump_intent = $this->read_json_file($dump_intent_file); + $dump_hash = hash_file('sha256', $sql_file); + if ( + ( $dump_intent['create_table_query'] ?? false ) !== true + || !is_string($dump_hash) + ) { + throw new RuntimeException( + 'Reprint could not confirm that db.sql finished downloading.', + ); + } + $this->write_json_file( + wp_join_unix_paths( + $this->pull_state_directory, + self::DATABASE_DUMP_RECORD_FILE, + ), + [ + 'sha256' => $dump_hash, + 'create_table_query' => true, + ], + ); + } + + // Mark the completed bytes before removing the download intent. If + // the process stops between these writes, the matching dump record + // still classifies db.sql and the leftover intent is harmless. + $this->get_state()->sql_bytes = null; $this->get_state()->active_resumable_command->completion_state = "complete"; $this->save_state(); - if (in_array($this->sql_output_mode, ['stdout', 'mysql'], true)) { + if ($this->sql_output_mode === 'file') { + if (file_exists($dump_intent_file)) { + @unlink($dump_intent_file); + } + } else { $this->write_json_file( wp_join_unix_paths( $this->pull_state_directory, @@ -3999,7 +4094,7 @@ private function run_db_domains(): void $sql_handle = fopen($sql_file, "r"); if (!$sql_handle) { - throw new RuntimeException("Cannot open SQL file: {$sql_file}"); + throw new RuntimeException("Reprint could not open db.sql for reading: {$sql_file}"); } try { @@ -5723,6 +5818,30 @@ public function run_db_apply(array $options): void "db.sql not found in {$this->state_dir}. Run db-pull first.", ); } + $sql_hash = hash_file('sha256', $sql_file); + if (!is_string($sql_hash)) { + // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- The path is the caller-selected CLI state directory. + throw new RuntimeException("Reprint could not read db.sql: {$sql_file}"); + } + $dump_record = $this->read_json_file(wp_join_unix_paths( + $this->pull_state_directory, + self::DATABASE_DUMP_RECORD_FILE, + )); + $is_confirmed_replacement_dump = + ( $dump_record['create_table_query'] ?? false ) === true + && hash_equals((string) ( $dump_record['sha256'] ?? '' ), $sql_hash); + $has_unconfirmed_download_intent = + is_file(wp_join_unix_paths( + $this->pull_state_directory, + self::DATABASE_DUMP_INTENT_FILE, + )) + && !$is_confirmed_replacement_dump; + if ($has_unconfirmed_download_intent) { + throw new RuntimeException( + 'db.sql is still downloading. Finish db-pull or run db-pull --abort ' . + 'before db-apply.', + ); + } // If --new-site-url is provided, derive the source origin from the // export URL and add an implicit --rewrite-url mapping. @@ -5894,7 +6013,7 @@ public function run_db_apply(array $options): void }); $sql_handle = fopen($sql_file, "r"); if (!$sql_handle) { - throw new RuntimeException("Cannot open SQL file: {$sql_file}"); + throw new RuntimeException("Reprint could not open db.sql for reading: {$sql_file}"); } $sql_file_size = filesize($sql_file); @@ -7621,34 +7740,64 @@ private function fetch_sql(): void if ($mode === "file") { $sql_file = wp_join_unix_paths($this->state_dir, "db.sql"); - // Crash recovery: if SQL file is larger than expected, truncate it. - // This happens if we crashed after writing but before saving the new cursor. + // Remove bytes written after the last saved source position before + // appending the next part of the dump. $tracked_bytes = $this->get_state()->sql_bytes ?? null; - if ($tracked_bytes !== null && file_exists($sql_file)) { + if ($cursor !== null) { + if (!is_int($tracked_bytes) || $tracked_bytes < 0 || !file_exists($sql_file)) { + throw new RuntimeException( + 'db-pull cannot continue because db.sql is missing or Reprint did not save ' . + 'its size. Run db-pull --abort, then start db-pull again.', + ); + } $actual_size = filesize($sql_file); + if ($actual_size === false || $actual_size < $tracked_bytes) { + throw new RuntimeException( + sprintf( + 'db-pull cannot continue because db.sql has %d bytes, but Reprint had saved ' . + 'a size of %d bytes. Run db-pull --abort, then start db-pull again.', + $actual_size === false ? 0 : $actual_size, + $tracked_bytes, + ), + ); + } if ($actual_size > $tracked_bytes) { $this->audit_log( sprintf( - "CRASH RECOVERY | Truncating db.sql from %d to %d bytes", + "RESUME | Truncating db.sql from %d to %d bytes", $actual_size, $tracked_bytes, ), true, ); $handle = fopen($sql_file, "r+"); - if ($handle) { - ftruncate($handle, $tracked_bytes); - fclose($handle); + if (!$handle || !ftruncate($handle, $tracked_bytes)) { + if ($handle) { + fclose($handle); + } + throw new RuntimeException( + 'Reprint could not remove unfinished bytes from db.sql. ' . + 'Run db-pull --abort, then start db-pull again.', + ); } + fclose($handle); } } - $sql_bytes_written = file_exists($sql_file) ? filesize($sql_file) : 0; + if ($cursor === null) { + $sql_bytes_written = 0; + } else { + $existing_size = filesize($sql_file); + if ($existing_size === false) { + throw new RuntimeException("Reprint could not read the size of db.sql: {$sql_file}"); + } + $sql_bytes_written = $existing_size; + } // Open in write mode if no cursor (starting fresh), append mode if resuming $sql_handle = fopen($sql_file, $cursor ? "a" : "w"); if (!$sql_handle) { - throw new RuntimeException("Cannot open SQL file: {$sql_file}"); + throw new RuntimeException("Reprint could not open db.sql for writing: {$sql_file}"); } } elseif ($mode === "stdout") { @@ -7739,6 +7888,11 @@ private function fetch_sql(): void try { while (!$complete) { $params = $this->get_tuned_params("sql_chunk"); + // A completed file dump may be replayed from byte zero only + // when every dumped base table replaces its target table. + if ($mode === 'file') { + $params['create_table_query'] = true; + } $url = $this->build_url("sql_chunk", $cursor, $params); $context = new StreamingContext(); @@ -7767,39 +7921,7 @@ private function fetch_sql(): void pcntl_signal_dispatch(); } - $cursor = $chunk["headers"]["x-cursor"] ?? $cursor; - - // File output saves both the SQL bytes and their source cursor. - // stdout and MySQL keep the cursor only for retries in this process. - $chunks_since_save++; - if ( - $mode === 'file' - && $chunks_since_save >= self::SAVE_STATE_EVERY_N_CHUNKS - && $sql_buffer === "" - ) { - if ($sql_handle) { - fflush($sql_handle); - } - $this->get_state()->active_resumable_command->remote_cursor = $cursor; - $this->get_state()->sql_bytes = $sql_bytes_written; - $this->get_state()->sql_statements_counted = $sql_statements_counted; - $this->save_state(); - $chunks_since_save = 0; - - // Also persist discovered domains so they survive crashes. - // On resume, the SQL download picks up from the cursor, - // skipping already-downloaded data — so domains from that - // earlier data would be lost without periodic saves. - if ($domain_collector) { - $domains = $domain_collector->get_domains(); - if (!empty($domains)) { - file_put_contents( - $domains_file, - json_encode($domains, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . "\n", - ); - } - } - } + $chunk_cursor = $chunk["headers"]["x-cursor"] ?? $cursor; $chunk_type = $chunk["headers"]["x-chunk-type"] ?? ""; @@ -7934,6 +8056,39 @@ private function fetch_sql(): void } elseif ($chunk_type === "error") { $this->handle_error_chunk($chunk, "sql", $context); } + + // The cursor names the part just processed. Only file output + // persists it after flushing the matching bytes. Direct output + // retains it in memory for source retries in this process. + $cursor = $chunk_cursor; + $chunks_since_save++; + if ( + $mode === 'file' + && $chunks_since_save >= self::SAVE_STATE_EVERY_N_CHUNKS + && $sql_buffer === "" + ) { + if ($sql_handle) { + if (!fflush($sql_handle)) { + throw new RuntimeException('Reprint could not finish writing db.sql before saving progress.'); + } + } + // Persist discovered domains before the cursor because + // resume skips SQL from before that cursor. + if ($domain_collector) { + $domains = $domain_collector->get_domains(); + if (!empty($domains)) { + file_put_contents( + $domains_file, + json_encode($domains, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . "\n", + ); + } + } + $this->get_state()->active_resumable_command->remote_cursor = $cursor; + $this->get_state()->sql_bytes = $sql_bytes_written; + $this->get_state()->sql_statements_counted = $sql_statements_counted; + $this->save_state(); + $chunks_since_save = 0; + } }; $cursor_before = $cursor; @@ -7966,11 +8121,15 @@ private function fetch_sql(): void $context->response_stats ?? [], ); + // Save the file size and source position together. if ($sql_handle) { - fflush($sql_handle); + if (!fflush($sql_handle)) { + throw new RuntimeException('Reprint could not finish writing db.sql before saving progress.'); + } $this->get_state()->active_resumable_command->remote_cursor = $cursor; - // Clear sql_bytes when complete, otherwise save current position. - $this->get_state()->sql_bytes = $complete ? null : $sql_bytes_written; + // Keep the final file size until run_db_sync records overall + // completion; that last state write may itself be interrupted. + $this->get_state()->sql_bytes = $sql_bytes_written; $this->save_state(); } } @@ -10917,12 +11076,16 @@ private function reset_state(): void /** Removes the files that describe a database pull. */ public function clear_database_pull_records(): void { - $path = wp_join_unix_paths( - $this->pull_state_directory, + foreach ([ + self::DATABASE_DUMP_INTENT_FILE, + self::DATABASE_DUMP_RECORD_FILE, self::DATABASE_PULL_OUTPUT_STATUS_FILE, - ); - if (file_exists($path) && !unlink($path)) { - throw new RuntimeException('Cannot delete the saved database output status.'); + ] as $filename) { + $path = wp_join_unix_paths($this->pull_state_directory, $filename); + if (file_exists($path) && !unlink($path)) { + // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- basename is one of the fixed record names above. + throw new RuntimeException('Cannot delete saved database pull file: ' . basename($path)); + } } } @@ -12607,7 +12770,8 @@ function _cli_option_usage(array $def): string "description" => "Indexes remote tables, then streams the full SQL dump into\n" . "--state-dir/db.sql (default), to stdout, or directly into a\n" . - "MySQL connection. Source interruptions retry within this process.\n" . + "MySQL connection. File output resumes from its saved cursor. Source\n" . + "interruptions retry within this process.\n" . "If that process stops, Reprint cannot tell how much SQL reached stdout or MySQL.\n" . "Reset or restore the destination, then abort and restart as file output.\n" . "Discovered domains are cached for later use by db-apply.\n", @@ -12647,7 +12811,7 @@ function _cli_option_usage(array $def): string "short" => "Print local pull metadata for host integrations as JSON", "usage" => "reprint pull-metadata --state-dir=DIR", "description" => - "Prints pull lifecycle, artifact availability, and source-site\n" . + "Prints pull progress, available downloads, and source-site\n" . "metadata for host integrations. The remote Reprint API URL selects\n" . "the state; no network calls are made.\n", "extra" => diff --git a/packages/reprint-client/src/lib/pull/class-pull.php b/packages/reprint-client/src/lib/pull/class-pull.php index 2c770c8a..83cad865 100644 --- a/packages/reprint-client/src/lib/pull/class-pull.php +++ b/packages/reprint-client/src/lib/pull/class-pull.php @@ -123,7 +123,7 @@ public function run(array $options): void * Handle --abort for high-level pull commands. * * File pipelines keep downloaded site files in place. The database - * pipeline removes stale database artifacts so the next pull-db fetches + * pipeline removes old database files so the next pull-db fetches * and applies a fresh dump. */ public function abort(string $command = 'pull'): void @@ -142,7 +142,7 @@ public function abort(string $command = 'pull'): void $label = $command === 'pull' ? 'Pull' : $command; $message = "{$label} state cleared."; $message .= $command === 'pull-db' - ? " Database artifacts will be downloaded again." + ? " Database files will be downloaded again." : " Downloaded files are preserved."; $this->progress->show_lifecycle_line("{$message}\n"); $this->client->output_progress([ @@ -312,7 +312,7 @@ private function run_pipeline( } } } elseif ($state_command === 'db-pull' && in_array('db-pull', $stages, true)) { - // Discard database dump artifacts from any previous runs. + // Discard database files from previous runs. $state = $this->client->get_state(); $state->active_resumable_command->command_name = null; $state->active_resumable_command->completion_state = null; @@ -325,9 +325,17 @@ private function run_pipeline( "{$state_dir}/db.sql", "{$state_dir}/db-tables.jsonl", "{$pull_state_directory}/domains.json", + "{$pull_state_directory}/sql-stats.json", ] as $path) { if (file_exists($path)) { - @unlink($path); + if (!unlink($path)) { + // phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- The path belongs to the caller-selected state directory. + throw new RuntimeException( + "Reprint could not delete pull file: {$path}. " . + "Check its permissions, then try again.", + ); + // phpcs:enable WordPress.Security.EscapeOutput.ExceptionNotEscaped + } } } $this->client->clear_database_pull_records(); @@ -831,11 +839,19 @@ private function prepare_repull(string $command): void $paths[] = wp_join_unix_paths($state_dir, 'db.sql'); $paths[] = wp_join_unix_paths($state_dir, 'db-tables.jsonl'); $paths[] = wp_join_unix_paths($pull_state_directory, 'domains.json'); + $paths[] = wp_join_unix_paths($pull_state_directory, 'sql-stats.json'); } foreach ($paths as $path) { if (file_exists($path)) { - @unlink($path); + if (!unlink($path)) { + // phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- The path belongs to the caller-selected state directory. + throw new RuntimeException( + "Reprint could not delete pull file: {$path}. " . + "Check its permissions, then try again.", + ); + // phpcs:enable WordPress.Security.EscapeOutput.ExceptionNotEscaped + } } } if ($reset_db_state) { diff --git a/tests/Import/DatabaseCommandRestartTest.php b/tests/Import/DatabaseCommandRestartTest.php new file mode 100644 index 00000000..717435e3 --- /dev/null +++ b/tests/Import/DatabaseCommandRestartTest.php @@ -0,0 +1,618 @@ +root = sys_get_temp_dir() . '/reprint-database-restart-' . uniqid('', true); + mkdir($this->root . '/state', 0755, true); + mkdir($this->root . '/files', 0755, true); + } + + protected function tearDown(): void + { + foreach ($this->childPids as $childPid) { + $waitResult = pcntl_waitpid($childPid, $status, WNOHANG); + if ($waitResult === 0 && function_exists('posix_kill') && defined('SIGKILL')) { + posix_kill($childPid, SIGKILL); + pcntl_waitpid($childPid, $status); + } + } + $this->removeTree($this->root); + parent::tearDown(); + } + + public function testDbPullContinuesAfterExitCodeTwo(): void + { + if (!function_exists('pcntl_fork')) { + $this->markTestSkipped('The local streaming endpoint requires pcntl.'); + } + + $sql = "SELECT 2;\n"; + [$remoteUrl, $serverPid] = $this->startOneResponseServer(function (array $query) use ($sql): string { + $this->assertSame('sql_chunk', $query['endpoint'] ?? null); + $this->assertSame('saved-sql-cursor', base64_decode($query['cursor'] ?? '', true)); + return $this->multipartResponse([ + [ + 'headers' => [ + 'X-Chunk-Type' => 'sql', + 'X-Query-Complete' => '1', + 'X-Cursor' => base64_encode('final-sql-cursor'), + ], + 'body' => $sql, + ], + [ + 'headers' => [ + 'X-Chunk-Type' => 'completion', + 'X-Status' => 'complete', + ], + 'body' => '', + ], + ]); + }); + + $client = $this->newClient($remoteUrl); + $this->writeReplacementDumpIntent($client); + $state = $client->get_state(); + $state->active_resumable_command->command_name = 'db-pull'; + $state->active_resumable_command->completion_state = 'partial'; + $state->active_resumable_command->current_stage = 'sql'; + $state->active_resumable_command->remote_cursor = base64_encode('saved-sql-cursor'); + $state->sql_bytes = strlen("SELECT 1;\n"); + $state->sql_output = 'file'; + $client->save_state(); + file_put_contents($this->root . '/state/db.sql', "SELECT 1;\n"); + + try { + $result = $this->runCli([ + 'db-pull', + $remoteUrl, + '--state-dir=' . $this->root . '/state', + '--fs-root=' . $this->root . '/files', + '--progress=jsonl', + ]); + } finally { + pcntl_waitpid($serverPid, $serverStatus); + } + + $this->assertSame(0, $result['exit'], $result['output']); + $this->assertStringContainsString('"event":"resuming"', $result['output']); + $this->assertSame("SELECT 1;\nSELECT 2;\n", file_get_contents($this->root . '/state/db.sql')); + $dumpRecord = json_decode( + (string) file_get_contents($client->pull_state_directory . '/database-dump.json'), + true, + ); + $this->assertIsArray($dumpRecord); + $this->assertSame( + hash_file('sha256', $this->root . '/state/db.sql'), + $dumpRecord['sha256'] ?? null, + ); + $this->assertTrue($dumpRecord['create_table_query'] ?? false); + $this->assertTrue(pcntl_wifexited($serverStatus)); + $this->assertSame(0, pcntl_wexitstatus($serverStatus)); + } + + public function testDbPullCrashKeepsThePartNamedByItsSavedCursor(): void + { + if (!function_exists('pcntl_fork') || !function_exists('posix_kill') || !defined('SIGKILL')) { + $this->markTestSkipped('The process-death test requires pcntl and posix signals.'); + } + + $readyPath = $this->root . '/sql-stream-ready'; + $releasePath = $this->root . '/sql-stream-release'; + [$remoteUrl, $serverPid] = $this->startTwoResponseSqlServer($readyPath, $releasePath); + $client = $this->newClient($remoteUrl); + $this->writeReplacementDumpIntent($client); + $state = $client->get_state(); + $state->active_resumable_command->command_name = 'db-pull'; + $state->active_resumable_command->completion_state = 'in_progress'; + $state->active_resumable_command->current_stage = 'sql'; + $state->sql_output = 'file'; + $client->save_state(); + + [$process] = $this->startCli([ + 'db-pull', + $remoteUrl, + '--state-dir=' . $this->root . '/state', + '--fs-root=' . $this->root . '/files', + '--progress=jsonl', + ], true); + + $pullStatePath = $client->pull_state_directory . '/state.json'; + $expectedFirstResponse = $this->sqlStatements(1, 50); + $this->waitUntil(function () use ($readyPath, $pullStatePath, $expectedFirstResponse): bool { + if (!is_file($readyPath) || !is_file($pullStatePath)) { + return false; + } + $state = json_decode( (string) file_get_contents($pullStatePath), true ); + return base64_decode($state['active_resumable_command']['remote_cursor'] ?? '', true) === 'cursor-50' + && is_file($this->root . '/state/db.sql') + && filesize($this->root . '/state/db.sql') === strlen($expectedFirstResponse); + }, 'The first process did not store the cursor for SQL part 50.'); + + $status = proc_get_status($process); + $this->assertTrue($status['running']); + $this->assertTrue(posix_kill($status['pid'], SIGKILL)); + proc_close($process); + file_put_contents($releasePath, ''); + + $result = $this->runCli([ + 'db-pull', + $remoteUrl, + '--state-dir=' . $this->root . '/state', + '--fs-root=' . $this->root . '/files', + '--progress=jsonl', + ]); + pcntl_waitpid($serverPid, $serverStatus); + + $this->assertSame(0, $result['exit'], $result['output']); + $this->assertSame($this->sqlStatements(1, 60), file_get_contents($this->root . '/state/db.sql')); + $this->assertTrue(pcntl_wifexited($serverStatus)); + $this->assertSame(0, pcntl_wexitstatus($serverStatus)); + } + + public function testDbPullDropsCursorlessBytesBeforeSavingItsFirstBoundary(): void + { + if (!function_exists('pcntl_fork') || !function_exists('posix_kill') || !defined('SIGKILL')) { + $this->markTestSkipped('The process-death test requires pcntl and posix signals.'); + } + + $firstReadyPath = $this->root . '/first-sql-stream-ready'; + $firstReleasePath = $this->root . '/first-sql-stream-release'; + $secondReadyPath = $this->root . '/second-sql-stream-ready'; + $secondReleasePath = $this->root . '/second-sql-stream-release'; + [$remoteUrl, $serverPid] = $this->startThreeResponseSqlServer( + $firstReadyPath, + $firstReleasePath, + $secondReadyPath, + $secondReleasePath, + ); + $client = $this->newClient($remoteUrl); + $this->writeReplacementDumpIntent($client); + $state = $client->get_state(); + $state->active_resumable_command->command_name = 'db-pull'; + $state->active_resumable_command->completion_state = 'in_progress'; + $state->active_resumable_command->current_stage = 'sql'; + $state->sql_output = 'file'; + $client->save_state(); + + [$firstProcess] = $this->startCli([ + 'db-pull', + $remoteUrl, + '--state-dir=' . $this->root . '/state', + '--fs-root=' . $this->root . '/files', + '--progress=jsonl', + ], true); + + $pullStatePath = $client->pull_state_directory . '/state.json'; + $firstUnconfirmedBytes = $this->sqlStatements(1, 10); + $this->waitUntil(function () use ($firstReadyPath, $pullStatePath, $firstUnconfirmedBytes): bool { + if (!is_file($firstReadyPath) || !is_file($pullStatePath)) { + return false; + } + $state = json_decode( (string) file_get_contents($pullStatePath), true ); + return empty($state['active_resumable_command']['remote_cursor']) + && is_file($this->root . '/state/db.sql') + && filesize($this->root . '/state/db.sql') === strlen($firstUnconfirmedBytes); + }, 'The first process did not write bytes before its first saved cursor.'); + + $firstStatus = proc_get_status($firstProcess); + $this->assertTrue($firstStatus['running']); + $this->assertTrue(posix_kill($firstStatus['pid'], SIGKILL)); + proc_close($firstProcess); + file_put_contents($firstReleasePath, ''); + + [$secondProcess] = $this->startCli([ + 'db-pull', + $remoteUrl, + '--state-dir=' . $this->root . '/state', + '--fs-root=' . $this->root . '/files', + '--progress=jsonl', + ], true); + + $secondCheckpointState = null; + $this->waitUntil(function () use ( + $secondProcess, + $secondReadyPath, + $pullStatePath, + &$secondCheckpointState + ): bool { + if (!is_file($secondReadyPath) || !is_file($pullStatePath)) { + return false; + } + if (!proc_get_status($secondProcess)['running']) { + return false; + } + $state = json_decode( (string) file_get_contents($pullStatePath), true ); + if (base64_decode($state['active_resumable_command']['remote_cursor'] ?? '', true) !== 'cursor-50') { + return false; + } + $secondCheckpointState = $state; + return true; + }, 'The second process did not save its first SQL cursor.'); + + $secondStatus = proc_get_status($secondProcess); + $this->assertTrue($secondStatus['running']); + $this->assertTrue(posix_kill($secondStatus['pid'], SIGKILL)); + proc_close($secondProcess); + file_put_contents($secondReleasePath, ''); + + $expectedCheckpointBytes = $this->sqlStatements(1, 50); + $this->assertSame(strlen($expectedCheckpointBytes), $secondCheckpointState['sql_bytes']); + $this->assertSame(strlen($expectedCheckpointBytes), filesize($this->root . '/state/db.sql')); + + $result = $this->runCli([ + 'db-pull', + $remoteUrl, + '--state-dir=' . $this->root . '/state', + '--fs-root=' . $this->root . '/files', + '--progress=jsonl', + ]); + pcntl_waitpid($serverPid, $serverStatus); + + $this->assertSame(0, $result['exit'], $result['output']); + $this->assertSame($this->sqlStatements(1, 60), file_get_contents($this->root . '/state/db.sql')); + $this->assertTrue(pcntl_wifexited($serverStatus)); + $this->assertSame(0, pcntl_wexitstatus($serverStatus)); + } + + private function newClient(string $remoteUrl): \ImportClient + { + $client = new \ImportClient( + $remoteUrl, + $this->root . '/state', + $this->root . '/files' + ); + $client->get_state()->set_preflight_record([ + 'http_code' => 200, + 'data' => ['ok' => true], + ]); + $client->save_state(); + return $client; + } + + private function writeReplacementDumpIntent(\ImportClient $client): void + { + file_put_contents( + $client->pull_state_directory . '/database-dump.intent', + json_encode(['create_table_query' => true], JSON_THROW_ON_ERROR), + ); + } + + /** @return array{exit:int,stdout:string,stderr:string,output:string} */ + private function runCli(array $arguments): array + { + [$process, $pipes] = $this->startCli($arguments); + $stdout = stream_get_contents($pipes[1]); + $stderr = stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + $exit = proc_close($process); + return [ + 'exit' => $exit, + 'stdout' => (string) $stdout, + 'stderr' => (string) $stderr, + 'output' => (string) $stdout . (string) $stderr, + ]; + } + + /** @return array{0:resource,1:array} */ + private function startCli(array $arguments, bool $discardOutput = false): array + { + $descriptors = $discardOutput + ? [['pipe', 'r'], ['file', '/dev/null', 'w'], ['file', '/dev/null', 'w']] + : [['pipe', 'r'], ['pipe', 'w'], ['pipe', 'w']]; + $process = proc_open( + array_merge([PHP_BINARY, __DIR__ . '/../../packages/reprint-client/bin/reprint-client'], $arguments), + $descriptors, + $pipes, + $this->root + ); + $this->assertIsResource($process); + fclose($pipes[0]); + return [$process, $pipes]; + } + + /** @return array{0:string,1:int} */ + private function startOneResponseServer(callable $response, int $acceptTimeout = 10): array + { + $listener = stream_socket_server('tcp://127.0.0.1:0', $errorNumber, $errorMessage); + $this->assertIsResource($listener, $errorMessage); + $address = stream_socket_get_name($listener, false); + $this->assertIsString($address); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + $connection = stream_socket_accept($listener, $acceptTimeout); + if (!is_resource($connection)) { + exit(2); + } + $request = $this->readHttpRequest($connection); + parse_str( (string) parse_url($request['target'], PHP_URL_QUERY), $query ); + fwrite($connection, $response($query)); + fclose($connection); + fclose($listener); + exit(0); + } + fclose($listener); + $this->childPids[] = $pid; + return ['http://' . $address . '/export', $pid]; + } + + /** @return array{0:string,1:int} */ + private function startTwoResponseSqlServer(string $readyPath, string $releasePath): array + { + $listener = stream_socket_server('tcp://127.0.0.1:0', $errorNumber, $errorMessage); + $this->assertIsResource($listener, $errorMessage); + $address = stream_socket_get_name($listener, false); + $this->assertIsString($address); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + $first = stream_socket_accept($listener, 10); + if (!is_resource($first)) { + exit(2); + } + $firstRequest = $this->readHttpRequest($first); + parse_str( (string) parse_url($firstRequest['target'], PHP_URL_QUERY), $firstQuery ); + if ( ( $firstQuery['create_table_query'] ?? null ) !== '1' ) { + exit(3); + } + fwrite($first, $this->multipartResponseHeaders('restart-boundary')); + for ($part = 1; $part <= 50; $part++) { + fwrite($first, $this->multipartPart('restart-boundary', [ + 'X-Chunk-Type' => 'sql', + 'X-Query-Complete' => '1', + 'X-Cursor' => base64_encode('cursor-' . $part), + ], sprintf("SELECT %d;\n", $part))); + fflush($first); + } + file_put_contents($readyPath, ''); + while (!is_file($releasePath)) { + usleep(10000); + } + fclose($first); + + $second = stream_socket_accept($listener, 10); + if (!is_resource($second)) { + exit(4); + } + $request = $this->readHttpRequest($second); + parse_str( (string) parse_url($request['target'], PHP_URL_QUERY), $query ); + if (base64_decode($query['cursor'] ?? '', true) !== 'cursor-50') { + exit(5); + } + if ( ( $query['create_table_query'] ?? null ) !== '1' ) { + exit(6); + } + $parts = []; + for ($part = 51; $part <= 60; $part++) { + $parts[] = [ + 'headers' => [ + 'X-Chunk-Type' => 'sql', + 'X-Query-Complete' => '1', + 'X-Cursor' => base64_encode('cursor-' . $part), + ], + 'body' => sprintf("SELECT %d;\n", $part), + ]; + } + $parts[] = [ + 'headers' => [ + 'X-Chunk-Type' => 'completion', + 'X-Status' => 'complete', + ], + 'body' => '', + ]; + fwrite($second, $this->multipartResponse($parts, 'restart-boundary')); + fclose($second); + fclose($listener); + exit(0); + } + fclose($listener); + $this->childPids[] = $pid; + return ['http://' . $address . '/export', $pid]; + } + + /** @return array{0:string,1:int} */ + private function startThreeResponseSqlServer( + string $firstReadyPath, + string $firstReleasePath, + string $secondReadyPath, + string $secondReleasePath + ): array { + $listener = stream_socket_server('tcp://127.0.0.1:0', $errorNumber, $errorMessage); + $this->assertIsResource($listener, $errorMessage); + $address = stream_socket_get_name($listener, false); + $this->assertIsString($address); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + $first = stream_socket_accept($listener, 10); + if (!is_resource($first)) { + exit(2); + } + $firstRequest = $this->readHttpRequest($first); + parse_str( (string) parse_url($firstRequest['target'], PHP_URL_QUERY), $firstQuery ); + if (isset($firstQuery['cursor'])) { + exit(3); + } + if ( ( $firstQuery['create_table_query'] ?? null ) !== '1' ) { + exit(4); + } + fwrite($first, $this->multipartResponseHeaders('cursorless-boundary')); + for ($part = 1; $part <= 10; $part++) { + fwrite($first, $this->multipartPart('cursorless-boundary', [ + 'X-Chunk-Type' => 'sql', + 'X-Query-Complete' => '1', + 'X-Cursor' => base64_encode('unconfirmed-' . $part), + ], sprintf("SELECT %d;\n", $part))); + fflush($first); + } + file_put_contents($firstReadyPath, ''); + while (!is_file($firstReleasePath)) { + usleep(10000); + } + fclose($first); + + $second = stream_socket_accept($listener, 10); + if (!is_resource($second)) { + exit(5); + } + $secondRequest = $this->readHttpRequest($second); + parse_str( (string) parse_url($secondRequest['target'], PHP_URL_QUERY), $secondQuery ); + if (isset($secondQuery['cursor'])) { + exit(6); + } + if ( ( $secondQuery['create_table_query'] ?? null ) !== '1' ) { + exit(7); + } + fwrite($second, $this->multipartResponseHeaders('cursorless-boundary')); + for ($part = 1; $part <= 50; $part++) { + fwrite($second, $this->multipartPart('cursorless-boundary', [ + 'X-Chunk-Type' => 'sql', + 'X-Query-Complete' => '1', + 'X-Cursor' => base64_encode('cursor-' . $part), + ], sprintf("SELECT %d;\n", $part))); + fflush($second); + } + file_put_contents($secondReadyPath, ''); + while (!is_file($secondReleasePath)) { + usleep(10000); + } + fclose($second); + + $third = stream_socket_accept($listener, 10); + if (!is_resource($third)) { + exit(8); + } + $thirdRequest = $this->readHttpRequest($third); + parse_str( (string) parse_url($thirdRequest['target'], PHP_URL_QUERY), $thirdQuery ); + if (base64_decode($thirdQuery['cursor'] ?? '', true) !== 'cursor-50') { + exit(9); + } + if ( ( $thirdQuery['create_table_query'] ?? null ) !== '1' ) { + exit(10); + } + $parts = []; + for ($part = 51; $part <= 60; $part++) { + $parts[] = [ + 'headers' => [ + 'X-Chunk-Type' => 'sql', + 'X-Query-Complete' => '1', + 'X-Cursor' => base64_encode('cursor-' . $part), + ], + 'body' => sprintf("SELECT %d;\n", $part), + ]; + } + $parts[] = [ + 'headers' => [ + 'X-Chunk-Type' => 'completion', + 'X-Status' => 'complete', + ], + 'body' => '', + ]; + fwrite($third, $this->multipartResponse($parts, 'cursorless-boundary')); + fclose($third); + fclose($listener); + exit(0); + } + fclose($listener); + $this->childPids[] = $pid; + return ['http://' . $address . '/export', $pid]; + } + + /** @return array{target:string} */ + private function readHttpRequest($connection): array + { + stream_set_timeout($connection, 10); + $request = ''; + while (strpos($request, "\r\n\r\n") === false) { + $chunk = fread($connection, 8192); + if ($chunk === false || $chunk === '') { + break; + } + $request .= $chunk; + } + $requestLine = strtok($request, "\r\n"); + $parts = is_string($requestLine) ? explode(' ', $requestLine) : []; + return ['target' => $parts[1] ?? '']; + } + + private function multipartResponse(array $parts, string $boundary = 'database-restart'): string + { + $body = ''; + foreach ($parts as $part) { + $body .= $this->multipartPart($boundary, $part['headers'], $part['body']); + } + $body .= "--{$boundary}--\r\n"; + return $this->multipartResponseHeaders($boundary, strlen($body)) . $body; + } + + private function multipartResponseHeaders(string $boundary, ?int $contentLength = null): string + { + $headers = "HTTP/1.1 200 OK\r\n" + . "Content-Type: multipart/mixed; boundary={$boundary}\r\n" + . "Connection: close\r\n"; + if ($contentLength !== null) { + $headers .= "Content-Length: {$contentLength}\r\n"; + } + return $headers . "\r\n"; + } + + private function multipartPart(string $boundary, array $headers, string $body): string + { + $part = "--{$boundary}\r\nContent-Length: " . strlen($body) . "\r\n"; + foreach ($headers as $name => $value) { + $part .= "{$name}: {$value}\r\n"; + } + return $part . "\r\n{$body}\r\n"; + } + + private function sqlStatements(int $first, int $last): string + { + $sql = ''; + for ($statement = $first; $statement <= $last; $statement++) { + $sql .= sprintf("SELECT %d;\n", $statement); + } + return $sql; + } + + private function waitUntil(callable $condition, string $failure): void + { + for ($attempt = 0; $attempt < 1000; $attempt++) { + if ($condition()) { + return; + } + usleep(10000); + } + $this->fail($failure); + } + + private function removeTree(string $path): void + { + if (is_link($path) || is_file($path)) { + @unlink($path); + return; + } + if (!is_dir($path)) { + return; + } + foreach (scandir($path) ?: [] as $entry) { + if ($entry !== '.' && $entry !== '..') { + $this->removeTree($path . '/' . $entry); + } + } + @rmdir($path); + } +} diff --git a/tests/e2e/site-registry.json b/tests/e2e/site-registry.json index 5465384d..30e69497 100644 --- a/tests/e2e/site-registry.json +++ b/tests/e2e/site-registry.json @@ -143,8 +143,14 @@ "stdout-db-pull-process-death": { "port": 8127 }, + "file-db-pull-artifact-guards": { + "port": 8130 + }, "direct-mysql-db-pull-safety": { "port": 8131 + }, + "db-pull-index-interruption": { + "port": 8132 } } } diff --git a/tests/e2e/tests/import-54-db-pull-index-interruption.test.js b/tests/e2e/tests/import-54-db-pull-index-interruption.test.js new file mode 100644 index 00000000..f9db73b7 --- /dev/null +++ b/tests/e2e/tests/import-54-db-pull-index-interruption.test.js @@ -0,0 +1,177 @@ +/** + * Test 54b: Database pull index response interruption. + * + * The source exits after sending its table-stat parts but before sending the + * completion part. The first db-pull invocation must return exit code 2 and + * retain partial state; a later process must continue that db-pull lifecycle + * instead of silently replacing it with a new one. + */ +import { describe, it, beforeAll, afterAll } from 'vitest'; +import assert from 'node:assert/strict'; +import { existsSync, readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { + runImporter, createTempDir, cleanupTempDir, + getSiteUrl, getSiteSecret, getSiteDir, + writeTestHooks, removeTestHooks, + readHookState, clearHookState, pullStateDirectory, +} from '../lib/test-helpers.js'; +import { ensureSite } from '../lib/site-setup.js'; + +describe('Import: Database Pull Response Interruption', () => { + const site = 'db-pull-index-interruption'; + const hookState = `/srv/e2e-sites/.e2e-hook-state-${site}`; + let tempDir; + let savedDbIndexCursor; + + beforeAll(async () => { + await ensureSite(site); + tempDir = createTempDir('e2e-db-pull-index-interruption'); + clearHookState(site); + }); + + afterAll(() => { + removeTestHooks(site); + clearHookState(site); + cleanupTempDir(tempDir); + }); + + function importUrl() { + return `${getSiteUrl(site)}&directory=${getSiteDir(site)}`; + } + + it('first db-pull process exits partial when the db-index completion part is interrupted', () => { + writeTestHooks(site, [ + 'function test_hook_before_completion($status, $gz, $boundary) {', + " if (($_GET['endpoint'] ?? '') !== 'db_index') { return; }", + ` $state = file_exists('${hookState}')`, + ` ? json_decode(file_get_contents('${hookState}'), true)`, + ' : [];', + " $state['requests'] = ($state['requests'] ?? 0) + 1;", + " $state['request_cursor'][] = $_GET['cursor'] ?? null;", + ` file_put_contents('${hookState}', json_encode($state));`, + " if ($state['requests'] === 1) {", + " $progress = json_encode(['phase' => 'tables']);", + ' $gz->write(', + ' "--{$boundary}\\r\\n" .', + ' "Content-Type: application/json\\r\\n" .', + ' "Content-Length: " . strlen($progress) . "\\r\\n" .', + ' "X-Chunk-Type: progress\\r\\n" .', + ' "\\r\\n" .', + ' $progress . "\\r\\n"', + ' );', + ' $gz->write("--{$boundary}--\\r\\n");', + ' $gz->finish();', + ' exit(1);', + ' }', + '}', + ].join('\n')); + + const result = runImporter(importUrl(), tempDir, 'db-pull', { + secret: getSiteSecret(site), + autoResume: false, + }); + + assert.equal( + result.exitCode, + 2, + `Expected exit 2 after the interrupted db-index response\nstderr: ${result.stderr}\nstdout: ${result.stdout}`, + ); + assert.deepEqual( + readHookState(site), + { requests: 1, request_cursor: [null] }, + 'Expected the first db-index request to start without a cursor', + ); + + const state = JSON.parse( + readFileSync(join(pullStateDirectory(tempDir, importUrl()), 'state.json'), 'utf-8'), + ); + assert.equal( + state.active_resumable_command.completion_state, + 'partial', + 'Expected db-pull to retain a partial checkpoint', + ); + assert.equal( + state.active_resumable_command.current_stage, + 'db-index', + 'Expected db-pull to stop in its db-index stage', + ); + savedDbIndexCursor = state.active_resumable_command.remote_cursor; + assert.ok( + savedDbIndexCursor, + 'Expected the interrupted db-index response to save its table cursor', + ); + assert.equal(state.sql_output, 'file'); + assert.equal( + existsSync(join(pullStateDirectory(tempDir, importUrl()), 'database-dump.intent')), + true, + 'Expected current file output to retain its download intent', + ); + }); + + it('the next db-pull process continues the partial lifecycle and completes', () => { + const changedMode = runImporter(importUrl(), tempDir, 'db-pull', { + secret: getSiteSecret(site), + autoResume: false, + extraArgs: ['--sql-output=stdout'], + }); + assert.equal( + changedMode.exitCode, + 1, + `Expected output-mode drift to fail:\n${changedMode.stderr}\n${changedMode.stdout}`, + ); + assert.match( + `${changedMode.stderr}\n${changedMode.stdout}`, + /Cannot change --sql-output/, + ); + assert.equal( + readHookState(site).requests, + 1, + 'Output-mode drift requested another database-index response', + ); + + const result = runImporter(importUrl(), tempDir, 'db-pull', { + secret: getSiteSecret(site), + }); + assert.equal( + result.exitCode, + 0, + `Expected db-pull resume to complete\nstderr: ${result.stderr}\nstdout: ${result.stdout}`, + ); + + const hookState = readHookState(site); + assert.equal(hookState.requests, 2, 'Expected a second db-index request'); + assert.equal( + hookState.request_cursor[1], + savedDbIndexCursor, + 'Expected the second db-index request to carry the first process cursor', + ); + + const lines = readFileSync( + join(tempDir, 'db-tables.jsonl'), + 'utf-8', + ).trim().split('\n').filter(Boolean); + const tableNames = lines.map((line) => JSON.parse(line).name); + + assert.ok(tableNames.length > 0, 'Expected table rows after resume'); + assert.equal( + new Set(tableNames).size, + tableNames.length, + 'Expected each table exactly once after resume', + ); + + const state = JSON.parse( + readFileSync(join(pullStateDirectory(tempDir, importUrl()), 'state.json'), 'utf-8'), + ); + assert.equal( + state.active_resumable_command.completion_state, + 'complete', + 'Expected db-pull to complete after resuming db-index', + ); + assert.match( + result.stdout, + /"event":"resuming".*"command":"db-pull"/, + 'Expected the second process to report that it continued db-pull', + ); + }); +}); diff --git a/tests/e2e/tests/import-59-file-db-pull-artifact-guards.test.js b/tests/e2e/tests/import-59-file-db-pull-artifact-guards.test.js new file mode 100644 index 00000000..56b7e1e1 --- /dev/null +++ b/tests/e2e/tests/import-59-file-db-pull-artifact-guards.test.js @@ -0,0 +1,300 @@ +/** + * Test 59: File database-pull artifact guards. + * + * The first cases deliberately tamper with db.sql after a causal process stop + * to lock the refusal when the file is shorter than the saved size. Another case + * proves that an actual unfinished managed pull cannot feed its prefix to + * db-apply as a one-shot arbitrary dump. + */ +import { describe, it, beforeAll, beforeEach, afterEach, afterAll } from 'vitest'; +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { + existsSync, + readFileSync, + truncateSync, + unlinkSync, +} from 'node:fs'; +import { join } from 'node:path'; +import { setTimeout as sleep } from 'node:timers/promises'; +import { + runImporter, + createTempDir, + cleanupTempDir, + getSiteUrl, + getSiteSecret, + getSiteDir, + fsRootDir, + writeTestHooks, + removeTestHooks, + readHookState, + clearHookState, + pullStateDirectory, +} from '../lib/test-helpers.js'; +import { ensureSite } from '../lib/site-setup.js'; + +describe('Import: file db-pull artifact guards', { timeout: 300000 }, () => { + const site = 'file-db-pull-artifact-guards'; + const projectRoot = join(import.meta.dirname, '..', '..', '..'); + const importerPath = process.env.IMPORTER_PATH + || join(projectRoot, 'packages', 'reprint-client', 'bin', 'reprint-client'); + const phpBinary = process.env.PHP_BINARY || 'php'; + const partialPullArguments = [ + '--sql-fragments-start=1', + '--sql-fragments-min=1', + '--sql-fragments-max=1', + '--progress=jsonl', + ]; + let tempDir; + let activeProcess; + let activeExit; + + function importUrl() { + return `${getSiteUrl(site)}&directory=${getSiteDir(site)}`; + } + + beforeAll(async () => { + await ensureSite(site, { + files: 'none', + customDb: async (_dbName, connection) => { + for (let index = 1; index <= 40; index++) { + const table = `artifact_guard_${String(index).padStart(2, '0')}`; + await connection.query( + `CREATE TABLE \`${table}\` (` + + '`id` INT NOT NULL, `value` VARCHAR(64) NOT NULL, ' + + 'PRIMARY KEY (`id`)) ENGINE=InnoDB' + ); + await connection.query( + `INSERT INTO \`${table}\` (id, value) VALUES (?, ?)`, + [index, `artifact-guard-${index}`], + ); + } + }, + }); + }); + + beforeEach(() => { + tempDir = createTempDir('e2e-file-db-pull-artifact-guards'); + clearHookState(site); + writeTestHooks(site, [ + 'function test_hook_before_sql_batch(&$sql, $cursor) {', + ` $state_file = '/srv/e2e-sites/.e2e-hook-state-${site}';`, + ' $state = file_exists($state_file)', + ' ? json_decode(file_get_contents($state_file), true)', + ' : [];', + " $state['sql_batches'] = ($state['sql_batches'] ?? 0) + 1;", + " if ($state['sql_batches'] === 61) {", + " $state['pause_started'] = true;", + ' file_put_contents($state_file, json_encode($state));', + ' usleep(3000000);', + " $state['pause_finished'] = true;", + ' }', + ' file_put_contents($state_file, json_encode($state));', + '}', + ].join('\n')); + }); + + afterEach(async () => { + if ( + activeProcess + && activeProcess.exitCode === null + && activeProcess.signalCode === null + ) { + activeProcess.kill('SIGKILL'); + await activeExit; + } + activeProcess = null; + activeExit = null; + removeTestHooks(site); + clearHookState(site); + if (tempDir) { + cleanupTempDir(tempDir); + } + }); + + afterAll(() => { + removeTestHooks(site); + clearHookState(site); + }); + + async function stopFilePullAfterSavedCursor() { + const preflight = runImporter(importUrl(), tempDir, 'preflight', { + secret: getSiteSecret(site), + autoResume: false, + }); + assert.equal( + preflight.exitCode, + 0, + `preflight failed:\n${preflight.stderr}\n${preflight.stdout}`, + ); + + const output = { stdout: '', stderr: '' }; + activeProcess = spawn(phpBinary, [ + importerPath, + 'db-pull', + importUrl(), + `--state-dir=${tempDir}`, + `--fs-root=${fsRootDir(tempDir)}`, + `--secret=${getSiteSecret(site)}`, + ...partialPullArguments, + ], { + env: { ...process.env }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + activeProcess.stdout.setEncoding('utf8'); + activeProcess.stderr.setEncoding('utf8'); + activeProcess.stdout.on('data', chunk => { output.stdout += chunk; }); + activeProcess.stderr.on('data', chunk => { output.stderr += chunk; }); + activeExit = new Promise(resolve => { + activeProcess.once('exit', (code, signal) => resolve({ code, signal })); + }); + + const statePath = join(pullStateDirectory(tempDir, importUrl()), 'state.json'); + const sqlPath = join(tempDir, 'db.sql'); + const pauseDeadline = Date.now() + 60000; + let interruptedState = null; + while (Date.now() < pauseDeadline) { + if (activeProcess.exitCode !== null || activeProcess.signalCode !== null) { + const result = await activeExit; + assert.fail( + `db-pull exited before the SQL pause (${result.code}/${result.signal}):\n` + + output.stderr + output.stdout, + ); + } + + const hookState = readHookState(site); + if (hookState?.pause_started && existsSync(statePath) && existsSync(sqlPath)) { + const state = JSON.parse(readFileSync(statePath, 'utf8')); + const command = state.active_resumable_command; + if ( + command?.current_stage === 'sql' + && command.remote_cursor + && Number(state.sql_bytes || 0) > 0 + ) { + interruptedState = state; + break; + } + } + await sleep(20); + } + + assert.ok(interruptedState, 'db-pull did not save the file size before the pause'); + assert.equal(interruptedState.sql_output, 'file'); + assert.equal(activeProcess.kill('SIGKILL'), true); + const killed = await activeExit; + assert.equal(killed.code, null); + assert.equal(killed.signal, 'SIGKILL'); + activeProcess = null; + activeExit = null; + + const serverDeadline = Date.now() + 10000; + while (!readHookState(site)?.pause_finished && Date.now() < serverDeadline) { + await sleep(20); + } + assert.equal( + readHookState(site)?.pause_finished, + true, + 'source did not leave the pause after the client process died', + ); + await sleep(500); + + return { interruptedState, sqlPath }; + } + + it.each(['missing', 'shorter'])( + 'rejects a deliberately tampered %s db.sql before requesting the saved cursor', + async artifactState => { + const { interruptedState, sqlPath } = await stopFilePullAfterSavedCursor(); + const savedBytes = Number(interruptedState.sql_bytes); + if (artifactState === 'missing') { + unlinkSync(sqlPath); + } else { + assert.ok(savedBytes > 1, 'saved SQL size is too small to shorten'); + truncateSync(sqlPath, savedBytes - 1); + } + const sqlBatchesBeforeResume = readHookState(site).sql_batches; + + const resumed = runImporter(importUrl(), tempDir, 'db-pull', { + secret: getSiteSecret(site), + autoResume: false, + extraArgs: partialPullArguments, + }); + + assert.equal( + resumed.exitCode, + 1, + `Expected ${artifactState} db.sql to fail closed:\n` + + resumed.stderr + resumed.stdout, + ); + const hookState = readHookState(site); + assert.equal( + hookState.sql_batches, + sqlBatchesBeforeResume, + 'db-pull requested a suffix after deliberate local artifact tampering', + ); + assert.match(`${resumed.stderr}\n${resumed.stdout}`, /db\.sql/i); + }, + ); + + it('rejects db-apply while a managed file pull contains only a prefix', async () => { + await stopFilePullAfterSavedCursor(); + const targetPath = join(tempDir, 'must-not-be-created.sqlite'); + const sqlBatchesBeforeApply = readHookState(site).sql_batches; + + const applied = runImporter(importUrl(), tempDir, 'db-apply', { + secret: getSiteSecret(site), + autoResume: false, + extraArgs: [ + '--target-engine=sqlite', + `--target-sqlite-path=${targetPath}`, + '--target-db=wordpress', + '--progress=jsonl', + ], + }); + + assert.equal( + applied.exitCode, + 1, + `Expected db-apply to reject an unfinished managed dump:\n` + + applied.stderr + applied.stdout, + ); + assert.equal( + existsSync(targetPath), + false, + 'db-apply opened the target before rejecting the unfinished dump', + ); + assert.equal(readHookState(site).sql_batches, sqlBatchesBeforeApply); + assert.match(`${applied.stderr}\n${applied.stdout}`, /still being downloaded/i); + }); + + it('rejects an output-mode change and preserves the file resume', async () => { + const { sqlPath } = await stopFilePullAfterSavedCursor(); + + const changedMode = runImporter(importUrl(), tempDir, 'db-pull', { + secret: getSiteSecret(site), + autoResume: false, + extraArgs: ['--sql-output=stdout'], + }); + assert.equal( + changedMode.exitCode, + 1, + `Expected output-mode change to fail:\n${changedMode.stderr}${changedMode.stdout}`, + ); + assert.match(`${changedMode.stderr}\n${changedMode.stdout}`, /Cannot change --sql-output/); + + const resumed = runImporter(importUrl(), tempDir, 'db-pull', { + secret: getSiteSecret(site), + autoResume: false, + extraArgs: partialPullArguments, + }); + assert.equal( + resumed.exitCode, + 0, + `file db-pull did not resume after rejected mode change:\n` + + resumed.stderr + resumed.stdout, + ); + const sql = readFileSync(sqlPath, 'utf8'); + assert.match(sql, /artifact_guard_01/); + assert.match(sql, /artifact_guard_40/); + }); +});