diff --git a/lib/Db/DbDumper.php b/lib/Db/DbDumper.php new file mode 100644 index 0000000..6273552 --- /dev/null +++ b/lib/Db/DbDumper.php @@ -0,0 +1,234 @@ +, rows: int, bytes: int, sha256: string}>, provider: string} + */ + public function dump(Repository $repo, array &$stats): array { + $sm = $this->db->createSchemaManager(); + $tables = array_values(array_filter($sm->listTableNames(), fn (string $t) => str_starts_with($t, $this->prefix))); + sort($tables); + + $this->db->setTransactionIsolation(TransactionIsolationLevel::REPEATABLE_READ); + $this->db->beginTransaction(); + try { + if ($this->db->getDatabasePlatform() instanceof \Doctrine\DBAL\Platforms\PostgreSQLPlatform) { + $this->db->executeStatement('SET TRANSACTION READ ONLY'); + } + $result = []; + foreach ($tables as $table) { + $writer = $repo->blobWriter($stats); + $hash = hash_init('sha256'); + $rows = 0; + foreach ($this->tableLines($sm->introspectTable($table)) as $kind => $line) { + $writer->write($line . "\n"); + if ($kind === 'row') { + hash_update($hash, $line . "\n"); + $rows++; + } + } + $result[$table] = ['blobs' => $writer->finish(), 'rows' => $rows, 'bytes' => $writer->bytes(), 'sha256' => hash_final($hash)]; + } + } finally { + $this->db->rollBack(); + } + $repo->flushPacks(); + return ['tables' => $result, 'provider' => get_class($this->db->getDatabasePlatform())]; + } + + /** + * Replace the contents of every table in the dump. Tables that exist but are not in the dump + * are left alone (e.g. a newer app); the caller decides whether that is acceptable. + * + * @param array{tables: array, rows: int}>} $manifest + */ + public function restore(Repository $repo, array $manifest): int { + $sm = $this->db->createSchemaManager(); + $existing = array_flip($sm->listTableNames()); + $total = 0; + $this->db->beginTransaction(); + try { + foreach ($manifest['tables'] as $table => $info) { + if (!isset($existing[$table])) { + throw new \RuntimeException("Table $table from the backup does not exist here (different app versions?)"); + } + $this->db->executeStatement('DELETE FROM ' . $this->db->quoteIdentifier($table)); + $columns = null; + $binary = []; + $batch = []; + foreach ($repo->readLines($info['blobs']) as $line) { + if ($line === '') { + continue; + } + $data = json_decode($line, true, 512, JSON_THROW_ON_ERROR); + if ($columns === null) { + $columns = $data['columns']; + $binary = array_flip($data['binary']); + continue; + } + $row = []; + foreach ($columns as $i => $col) { + $value = $data[$i]; + $row[$col] = ($value !== null && isset($binary[$col])) ? base64_decode($value, true) : $value; + } + $batch[] = $row; + if (count($batch) >= 500) { + $total += $this->insertBatch($table, $batch, $binary); + $batch = []; + } + } + if ($batch !== []) { + $total += $this->insertBatch($table, $batch, $binary); + } + $this->resetSequences($table, $sm->introspectTable($table)); + } + $this->db->commit(); + } catch (\Throwable $e) { + $this->db->rollBack(); + throw $e; + } + return $total; + } + + /** Same checksum as dump(): sha256 over the row lines in primary-key order. */ + public function checksum(string $table): array { + $sm = $this->db->createSchemaManager(); + $hash = hash_init('sha256'); + $rows = 0; + foreach ($this->tableLines($sm->introspectTable($table)) as $kind => $line) { + if ($kind === 'row') { + hash_update($hash, $line . "\n"); + $rows++; + } + } + return ['rows' => $rows, 'sha256' => hash_final($hash)]; + } + + /** @return \Generator 'header' => line, then 'row' => line ... */ + private function tableLines(\Doctrine\DBAL\Schema\Table $table): \Generator { + $columns = array_values(array_map(fn ($c) => $c->getName(), $table->getColumns())); + $binary = []; + foreach ($table->getColumns() as $c) { + if ($c->getType() instanceof BlobType || $c->getType() instanceof BinaryType) { + $binary[] = $c->getName(); + } + } + $pk = $table->getPrimaryKey()?->getColumns() ?? []; + yield 'header' => json_encode(['table' => $table->getName(), 'columns' => array_values($columns), 'binary' => $binary, 'pk' => $pk], JSON_THROW_ON_ERROR); + + $binaryFlip = array_flip($binary); + $q = fn (string $c) => $this->db->quoteIdentifier($c); + $select = 'SELECT ' . implode(', ', array_map($q, $columns)) . ' FROM ' . $q($table->getName()); + + if ($pk === []) { + // Nextcloud requires primary keys; fall back to one query for odd tables. + $batches = [$this->db->fetchAllNumeric($select)]; + } else { + $batches = $this->keysetBatches($select, $pk, $columns); + } + foreach ($batches as $rows) { + foreach ($rows as $row) { + foreach ($columns as $i => $col) { + if (is_resource($row[$i])) { + $row[$i] = stream_get_contents($row[$i]); + } + if ($row[$i] !== null && isset($binaryFlip[$col])) { + $row[$i] = base64_encode((string)$row[$i]); + } elseif (is_int($row[$i]) || is_float($row[$i]) || is_bool($row[$i])) { + // Normalise: drivers differ in returning numbers as int or string. + $row[$i] = (string)(is_bool($row[$i]) ? (int)$row[$i] : $row[$i]); + } + } + yield 'row' => json_encode($row, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE | JSON_THROW_ON_ERROR); + } + } + } + + /** @return \Generator>> */ + private function keysetBatches(string $select, array $pk, array $columns): \Generator { + $q = fn (string $c) => $this->db->quoteIdentifier($c); + $order = ' ORDER BY ' . implode(', ', array_map($q, $pk)); + $pkIdx = array_map(fn ($c) => array_search($c, $columns, true), $pk); + $last = null; + while (true) { + $sql = $select; + $params = []; + if ($last !== null) { + // Row-value comparison (a, b) > (?, ?), supported by PostgreSQL, MySQL/MariaDB and SQLite. + $sql .= ' WHERE (' . implode(', ', array_map($q, $pk)) . ') > (' . implode(', ', array_fill(0, count($pk), '?')) . ')'; + $params = $last; + } + $rows = $this->db->fetchAllNumeric($sql . $order . ' LIMIT ' . self::BATCH, $params); + if ($rows === []) { + return; + } + yield $rows; + $end = end($rows); + $last = array_map(fn ($i) => $end[$i], $pkIdx); + if (count($rows) < self::BATCH) { + return; + } + } + } + + private function insertBatch(string $table, array $rows, array $binary): int { + foreach ($rows as $row) { + $types = []; + foreach ($row as $col => $_) { + $types[$col] = isset($binary[$col]) ? \Doctrine\DBAL\ParameterType::LARGE_OBJECT : \Doctrine\DBAL\ParameterType::STRING; + } + $this->db->insert($this->db->quoteIdentifier($table), array_combine(array_map(fn ($c) => $this->db->quoteIdentifier($c), array_keys($row)), $row), array_values($types)); + } + return count($rows); + } + + private function resetSequences(string $table, \Doctrine\DBAL\Schema\Table $schema): void { + if (!$this->db->getDatabasePlatform() instanceof \Doctrine\DBAL\Platforms\PostgreSQLPlatform) { + return; // MySQL/SQLite derive AUTO_INCREMENT from the data + } + foreach ($schema->getColumns() as $column) { + if ($column->getAutoincrement()) { + $this->db->executeStatement(sprintf( + "SELECT setval(pg_get_serial_sequence('%s', '%s'), COALESCE((SELECT MAX(%s) FROM %s), 0) + 1, false)", + $table, $column->getName(), $this->db->quoteIdentifier($column->getName()), $this->db->quoteIdentifier($table))); + } + } + } +} diff --git a/lib/Repository/BlobWriter.php b/lib/Repository/BlobWriter.php new file mode 100644 index 0000000..ffa7633 --- /dev/null +++ b/lib/Repository/BlobWriter.php @@ -0,0 +1,49 @@ + */ + private array $blobs = []; + private int $bytes = 0; + + /** @param \Closure(string): string $store stores one blob, returns its id */ + public function __construct( + private \Closure $store, + private int $blobSize = Repository::BLOB_SIZE, + ) { + } + + public function write(string $data): void { + $this->buffer .= $data; + $this->bytes += strlen($data); + while (strlen($this->buffer) >= $this->blobSize) { + $this->blobs[] = ($this->store)(substr($this->buffer, 0, $this->blobSize)); + $this->buffer = substr($this->buffer, $this->blobSize); + } + } + + /** @return list blob ids, in order */ + public function finish(): array { + if ($this->buffer !== '') { + $this->blobs[] = ($this->store)($this->buffer); + $this->buffer = ''; + } + return $this->blobs; + } + + public function bytes(): int { + return $this->bytes; + } +} diff --git a/lib/Repository/Repository.php b/lib/Repository/Repository.php index 223f7e5..6d51d3d 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -31,6 +31,7 @@ final class Repository { private StreamCipher $cipher; private BlobIndex $index; + private ?PackWriter $runPacks = null; private function __construct( private IBackend $backend, @@ -145,6 +146,53 @@ final class Repository { return $count; } + /** + * Start writing a generated stream (e.g. a table dump) as blobs. Call flushPacks() when the + * run is done, so the last pack is uploaded before anything references it. + */ + public function blobWriter(array &$stats): BlobWriter { + $this->runPacks ??= new PackWriter($this->backend, $this->cipher, $this->index); + $packs = $this->runPacks; + return new BlobWriter(function (string $data) use ($packs, &$stats): string { + return $this->storeBlob($data, $packs, $stats); + }); + } + + public function flushPacks(): void { + $this->runPacks?->flush(); + } + + /** + * Read a blob stream back as lines (lines may span blob boundaries). + * + * @param list $blobIds + * @return \Generator + */ + public function readLines(array $blobIds): \Generator { + $carry = ''; + foreach ($blobIds as $id) { + $data = $carry . $this->loadBlob($id); + $lines = explode("\n", $data); + $carry = array_pop($lines); + yield from $lines; + } + if ($carry !== '') { + yield $carry; + } + } + + /** Small encrypted objects (manifests). */ + public function putObject(string $path, string $data): void { + self::putString($this->backend, $path, $this->cipher->encryptString($data, 'object:' . $path)); + } + + public function getObject(string $path): string { + $fh = $this->backend->get($path); + $data = $this->cipher->decryptString((string)stream_get_contents($fh), 'object:' . $path); + fclose($fh); + return $data; + } + /** @return list */ public function listSnapshots(): array { return array_map('basename', $this->backend->list('snapshots')); diff --git a/spikes/03-db-dump.php b/spikes/03-db-dump.php new file mode 100644 index 0000000..8b00f58 --- /dev/null +++ b/spikes/03-db-dump.php @@ -0,0 +1,112 @@ +getInner(); +$prefix = \OCP\Server::get(\OCP\IConfig::class)->getSystemValueString('dbtableprefix', 'oc_'); +$dumper = new DbDumper($db, $prefix); +printf("database: %s, prefix %s\n", get_class($db->getDatabasePlatform()), $prefix); + +$tables = array_values(array_filter($db->createSchemaManager()->listTableNames(), fn ($t) => str_starts_with($t, $prefix))); +sort($tables); +$before = []; +foreach ($tables as $t) { + $before[$t] = $dumper->checksum($t); +} +printf("%d tables, %d rows\n", count($tables), array_sum(array_column($before, 'rows'))); + +$repo = Repository::init(new LocalBackend($repoDir), 'spike passphrase'); + +memory_reset_peak_usage(); +$stats = ['newBlobs' => 0, 'dupBlobs' => 0, 'uploaded' => 0]; +$t = microtime(true); +$manifest = $dumper->dump($repo, $stats); +$repo->putObject('db/1', json_encode($manifest, JSON_THROW_ON_ERROR)); +printf("dump 1: %.1fs, peak %.1f MiB, %s\n", microtime(true) - $t, memory_get_peak_usage() / 1048576, json_encode($stats)); +$dumpBytes = array_sum(array_column($manifest['tables'], 'bytes')); +printf("dump size (plain JSON lines): %.1f MiB\n", $dumpBytes / 1048576); +$check('checksums in dump match live tables', array_map(fn ($x) => $x['sha256'], $manifest['tables']) === array_map(fn ($x) => $x['sha256'], $before)); + +// Second dump without changes: blobs are reused. +$stats2 = ['newBlobs' => 0, 'dupBlobs' => 0, 'uploaded' => 0]; +$dumper->dump($repo, $stats2); +printf("dump 2 (no changes): %s\n", json_encode($stats2)); +$check('unchanged database: (almost) nothing new stored', $stats2['uploaded'] < 0.05 * max(1, $stats['uploaded'])); + +// Damage the database. +$db->executeStatement('DELETE FROM ' . $prefix . 'filecache WHERE path LIKE ?', ['files/Docs/%']); +$db->executeStatement('UPDATE ' . $prefix . 'appconfig SET configvalue = ? WHERE appid = ? AND configkey = ?', ['damaged', 'core', 'installedat']); +$db->insert($prefix . 'users', ['uid' => 'intruder', 'displayname' => 'Intruder', 'password' => 'x', 'uid_lower' => 'intruder']); +$changed = array_filter($tables, fn ($tb) => $dumper->checksum($tb)['sha256'] !== $before[$tb]['sha256']); +printf("tables changed by the damage: %s\n", implode(', ', $changed)); +$check('damage visible in checksums', count($changed) >= 3); + +// Restore. +memory_reset_peak_usage(); +$t = microtime(true); +$repo2 = Repository::open(new LocalBackend($repoDir), 'spike passphrase'); +$m = json_decode($repo2->getObject('db/1'), true, 512, JSON_THROW_ON_ERROR); +$rows = $dumper->restore($repo2, $m); +printf("restore: %d rows in %.1fs, peak %.1f MiB\n", $rows, microtime(true) - $t, memory_get_peak_usage() / 1048576); + +$mismatch = []; +foreach ($tables as $tb) { + $after = $dumper->checksum($tb); + if ($after !== $before[$tb]) { + $mismatch[] = $tb; + } +} +$check('after restore: every table identical to before', $mismatch === []); +if ($mismatch) { + echo 'mismatch: ' . implode(', ', $mismatch) . "\n"; +} + + +// Sequences must continue after the restored maximum (PostgreSQL). +if ($db->getDatabasePlatform() instanceof \Doctrine\DBAL\Platforms\PostgreSQLPlatform) { + $serials = $db->fetchAllAssociative("SELECT table_name, column_name, pg_get_serial_sequence(table_name, column_name) AS seq + FROM information_schema.columns WHERE table_schema = current_schema() AND table_name LIKE ? + AND pg_get_serial_sequence(table_name, column_name) IS NOT NULL", [$prefix . '%']); + $bad = []; + foreach ($serials as $s) { + $max = (int)$db->fetchOne(sprintf('SELECT COALESCE(MAX(%s), 0) FROM %s', $db->quoteIdentifier($s['column_name']), $db->quoteIdentifier($s['table_name']))); + $seq = $db->fetchAssociative('SELECT last_value, is_called FROM ' . $s['seq']); + $next = (int)$seq['last_value'] + ($seq['is_called'] ? 1 : 0); + if ($next <= $max) { + $bad[] = "{$s['table_name']}.{$s['column_name']} next=$next max=$max"; + } + } + $check(sprintf('all %d sequences continue above the restored maximum', count($serials)), $bad === []); + if ($bad) { + echo implode("\n", $bad), "\n"; + } +} + + +exec('rm -rf ' . escapeshellarg($repoDir)); +echo $ok ? "ALL PASS\n" : "FAILURES\n"; +exit($ok ? 0 : 1);