From c3b70cbca594ae8979a0f7030a82d4f2d09935fc Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 22 Sep 2026 17:52:28 -0600 Subject: [PATCH 1/4] fix(sqlite): seek to the positional resume point instead of replaying The positional cursor on cloudsync_payload_chunks stated its resume lower bound only inside (db_version>? OR (db_version=? AND seq>=?)). The two arms carry distinct parameters, so SQLite derives no range from the disjunction and cloudsync_changes' xBestIndex was offered an upper bound and a site filter but no lower bound. Every call re-read the window from the start, evaluating cloudsync_col_value() on each discarded row, which made a full drain quadratic in the number of chunks. State db_version>=? explicitly alongside the disjunction. The term is logically implied, so the same rows are selected; it exists so the constraint loop emits a lower bound into the generated inner SQL and the (db_version) index can seek. Draining a 188-chunk window locally: 6110ms -> 213ms, with per-chunk cost now constant in the window size rather than proportional to it. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 1 + src/sqlite/cloudsync_sqlite.c | 10 ++++++++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e123fa3f..e4c1938b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ### Fixed +- **SQLite: paging a chunked download no longer gets slower with every chunk.** The positional cursor on `cloudsync_payload_chunks` was meant to seek straight to where the previous call stopped, but it stated its resume point only inside `(db_version > ? OR (db_version = ? AND seq >= ?))`, whose two arms carry distinct parameters. SQLite does not derive a range from that, so the scan over `cloudsync_changes` ran with an upper bound only and re-read the window from the beginning on every call, discarding rows until it reached the resume point — making a full drain quadratic in the number of chunks, and long enough on a large tenant to hit a server-side deadline and never complete. An explicit `db_version >= ?` is now stated alongside the disjunction; it selects exactly the same rows and lets the scan seek. Locally, draining a 188-chunk window went from 6110 ms to 210 ms, and the per-chunk cost no longer depends on how large the window is. PostgreSQL was never affected. - **A row rewritten in one statement no longer keeps the old blocks of a shorter value.** Writing a whole row rewrites its block column from the first position, so any block past the end of the new value stayed stored. On SQLite `INSERT OR REPLACE` skips the old row's delete trigger unless `recursive_triggers` is on, so those leftovers kept their metadata and were delivered as content: replacing `AAA\nBBB\nCCC` with `ZZZ` left `ZZZ\nBBB\nCCC` on the peers and on a later local read. On PostgreSQL they carried no metadata, so peers were unaffected, but they stayed in the blocks table for the life of the row. Blocks the new value does not cover are now retired with it. - **SQLite: a payload whose commit fails no longer leaves its transaction open.** When `cloudsync_payload_apply` started the transaction itself and the commit then failed — a deferred foreign key violated at commit, or `SQLITE_BUSY` because a reader held the database — the transaction stayed open: the uncommitted rows remained visible on the connection and the next `BEGIN` failed. The failed transaction is now rolled back and the original error is returned. Changes from earlier source versions that were already committed are kept, the receive checkpoint does not move, and the rolled-back rows are no longer counted as applied, so delivering the payload again applies it. A transaction or savepoint opened by the caller is still left to the caller. - **PostgreSQL: applying a payload read from a table works at any savepoint depth.** Inside 126 or more savepoints, `SELECT cloudsync_payload_apply(payload) FROM some_table` still failed with `buffer pin ... is not owned by resource owner SubTransaction` (and a caught error at that depth could abort an assertion-enabled server): cloudsync recorded the caller's resource owner and memory context for at most 128 nesting levels, counting its own internal savepoints, and silently stopped restoring them beyond that. The fixed limit is gone; only PostgreSQL's own resource limits apply. diff --git a/src/sqlite/cloudsync_sqlite.c b/src/sqlite/cloudsync_sqlite.c index 25cdd7c3..a64ab1c8 100644 --- a/src/sqlite/cloudsync_sqlite.c +++ b/src/sqlite/cloudsync_sqlite.c @@ -1401,9 +1401,14 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const // (db_version, seq) >= (resume_dbv, resume_seq). char *sql; if (positional) { + // The redundant db_version>=? is what makes the resume a seek. The + // disjunction alone states the same bound, but its two arms carry distinct + // parameters, so SQLite cannot derive a range from it and cloudsync_changes' + // xBestIndex is offered no lower bound at all — leaving every call to replay + // the window from the start. See docs/internal/payload-chunks-resume-scan.md. sql = sqlite3_mprintf( "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq " - "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND " + "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND db_version>=? AND " "(db_version>? OR (db_version=? AND seq>=?)) ORDER BY db_version, seq ASC", site_op); } else { @@ -1421,7 +1426,8 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT); sqlite3_bind_int64(c->src, 3, resume_dbv); sqlite3_bind_int64(c->src, 4, resume_dbv); - sqlite3_bind_int64(c->src, 5, resume_seq); + sqlite3_bind_int64(c->src, 5, resume_dbv); + sqlite3_bind_int64(c->src, 6, resume_seq); } else { sqlite3_bind_int64(c->src, 1, since); sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT); From 5bac6e144a980ae81c77230a18ae5a4066f87ac6 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 22 Sep 2026 17:52:28 -0600 Subject: [PATCH 2/4] test: add a local benchmark for the positional chunk drain make chunk-bench times a real positional drain per chunk index, then replays the same resume points straight at cloudsync_changes in three SQL shapes and prints the idxStr each one produces, so whether a lower bound reaches xBestIndex is visible rather than inferred. It asserts every shape selects the identical row at each resume point. CI builds it along with the other test binaries, via the wildcard in TEST_SRC, but never runs it: the timings are machine-dependent and the shape of the curve, not the absolute numbers, is the result. Co-Authored-By: Claude Opus 5 (1M context) --- Makefile | 11 ++ test/chunk_bench.c | 258 +++++++++++++++++++++++++++++++++++++++------ 2 files changed, 237 insertions(+), 32 deletions(-) diff --git a/Makefile b/Makefile index aeec4229..4c119a14 100644 --- a/Makefile +++ b/Makefile @@ -357,6 +357,17 @@ sync-bench: $(TARGET) $(DIST_DIR)/sync_bench$(EXE) sync-bench-debug: $(MAKE) NETWORK_TRACE=1 sync-bench +# Time the positional /check drain on a locally generated window. Separate from +# the test targets because timings are machine-dependent; it needs no network. +chunk-bench: $(TARGET) $(DIST_DIR)/chunk_bench$(EXE) + @if [ -n "$(CHUNK_BENCH_ROWS)" ]; then export CHUNK_BENCH_ROWS="$(CHUNK_BENCH_ROWS)"; fi; \ + if [ -n "$(CHUNK_BENCH_ROW_BYTES)" ]; then export CHUNK_BENCH_ROW_BYTES="$(CHUNK_BENCH_ROW_BYTES)"; fi; \ + if [ -n "$(CHUNK_BENCH_TXNS)" ]; then export CHUNK_BENCH_TXNS="$(CHUNK_BENCH_TXNS)"; fi; \ + if [ -n "$(CHUNK_BENCH_REPEATS)" ]; then export CHUNK_BENCH_REPEATS="$(CHUNK_BENCH_REPEATS)"; fi; \ + if [ -n "$(CHUNK_BENCH_CHUNK_SIZE)" ]; then export CHUNK_BENCH_CHUNK_SIZE="$(CHUNK_BENCH_CHUNK_SIZE)"; fi; \ + if [ -n "$(CHUNK_BENCH_VERBOSE)" ]; then export CHUNK_BENCH_VERBOSE="$(CHUNK_BENCH_VERBOSE)"; fi; \ + ./$(DIST_DIR)/chunk_bench$(EXE) + OPENSSL_TARBALL = $(OPENSSL_DIR)/$(OPENSSL_VERSION).tar.gz $(OPENSSL_TARBALL): diff --git a/test/chunk_bench.c b/test/chunk_bench.c index 4321ac2f..010b6b52 100644 --- a/test/chunk_bench.c +++ b/test/chunk_bench.c @@ -2,28 +2,74 @@ // chunk_bench.c // cloudsync // -// Local-only benchmark for the positional /check drain: build a window of N -// chunks and time paging the whole window one chunk per call via the -// (resume_db_version, resume_seq, resume_frag_offset) cursor on -// cloudsync_payload_chunks. Reports wall time and per-chunk cost so the -// computational growth of the drain (currently O(N^2): each resume re-scans -// cloudsync_changes) can be tracked — e.g. to confirm a future indexed -// (db_version, seq) seek flattens it to O(N). +// Local-only benchmark for the positional /check drain. Two measurements, on +// one generated window: // -// Env: CHUNK_BENCH_ROWS (default 400), CHUNK_BENCH_ROW_BYTES (default 60000), -// CHUNK_BENCH_TXNS (default 1; rows split across this many db_versions), -// CHUNK_BENCH_REPEATS (default 5), CHUNK_BENCH_CHUNK_SIZE (default 262144). +// Phase 1 — drain the whole window one chunk per call through the +// (resume_db_version, resume_seq, resume_frag_offset) cursor on +// cloudsync_payload_chunks, timing every chunk. Latency that rises with the +// chunk index means each resume re-scans cloudsync_changes from the start of +// the window, making the drain O(N^2); flat latency means the resume seeks. +// +// Phase 2 — replay the same resume points straight against cloudsync_changes +// in two SQL shapes: the one the positional branch emits today, and the same +// clause plus the redundant AND-connected `db_version>=?` lower bound +// proposed in docs/internal/payload-chunks-resume-scan.md. Same rows, two +// plans, so the difference is the value of that proposal. +// +// CI builds this with the other test binaries but never runs it: the timings are +// machine-dependent. Run it by hand with `make chunk-bench`. +// +// Env: CHUNK_BENCH_ROWS (default 3000), CHUNK_BENCH_ROW_BYTES (default 8000), +// CHUNK_BENCH_TXNS (default = rows, i.e. one db_version per row; 1 is the +// degenerate single-version window where no lower bound can help), +// CHUNK_BENCH_REPEATS (default 3), CHUNK_BENCH_CHUNK_SIZE (default 262144), +// CHUNK_BENCH_VERBOSE (1 = print every chunk, not just deciles). // #include #include #include #include +#include #include #include "sqlite3.h" #define DB_PATH "dist/chunk-bench.sqlite" #define EXT_PATH "./dist/cloudsync" +#define MAX_POINTS 20000 +#define BUCKETS 10 + +// Every shape binds ?1=until, ?2=site_id, ?3/?4/?6=resume_db_version, ?5=resume_seq, +// and selects exactly the same rows. Only the spelling of the lower bound differs. + +// What the positional branch emits today (cloudsync_sqlite.c:1403-1414): the resume +// point lives entirely inside a disjunction whose two arms carry *distinct* +// parameters. Distinct parameters are what defeats the planner here — see below. +#define SHAPE_CURRENT \ + "SELECT db_version, seq FROM cloudsync_changes " \ + "WHERE db_version<=?1 AND site_id<>?2 AND (db_version>?3 OR (db_version=?4 AND seq>=?5)) " \ + "ORDER BY db_version, seq ASC LIMIT 1" + +// The doc's proposal: a redundant conjunct stating the lower bound where a virtual +// table's xBestIndex can see it, alongside the untouched disjunction. +#define SHAPE_PROPOSED \ + "SELECT db_version, seq FROM cloudsync_changes " \ + "WHERE db_version<=?1 AND site_id<>?2 AND db_version>=?6 AND (db_version>?3 OR (db_version=?4 AND seq>=?5)) " \ + "ORDER BY db_version, seq ASC LIMIT 1" + +// The same disjunction with one parameter reused across both arms. SQLite derives +// the common lower bound itself once it can see the two arms compare against the +// same value, so this needs no extra term — and it means the defect is a matter of +// parameter identity, not of disjunctions being opaque to xBestIndex. +#define SHAPE_REUSED_PARAM \ + "SELECT db_version, seq FROM cloudsync_changes " \ + "WHERE db_version<=?1 AND site_id<>?2 AND (db_version>?3 OR (db_version=?3 AND seq>=?5)) " \ + "ORDER BY db_version, seq ASC LIMIT 1" + +typedef struct { + int64_t dbv, seq, frag; +} resume_point; static double monotonic_ms(void) { struct timespec ts; @@ -50,9 +96,26 @@ static int db_exec(sqlite3 *db, const char *sql) { return rc; } -// Drain the whole window via the positional cursor, one chunk per query. Returns -// the chunk count and accumulates total payload bytes touched into *bytes. -static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) { +// Print the clause cloudsync_changesvtab_best_index hands the planner for one +// shape. This is the doc's verification step 3: whether the resume lower bound +// reaches the virtual table at all is visible right here, in the idxStr. +static void print_plan(sqlite3 *db, const char *label, const char *sql) { + char *eqp = sqlite3_mprintf("EXPLAIN QUERY PLAN %s", sql); + if (!eqp) return; + sqlite3_stmt *stmt = NULL; + if (sqlite3_prepare_v2(db, eqp, -1, &stmt, NULL) == SQLITE_OK) { + while (sqlite3_step(stmt) == SQLITE_ROW) + printf(" %-9s %s\n", label, sqlite3_column_text(stmt, 3)); + } + sqlite3_finalize(stmt); + sqlite3_free(eqp); +} + +// Drain the whole window via the positional cursor, one chunk per query, timing +// each call and recording the resume point it was issued with so phase 2 can +// replay exactly the same positions. +static int drain_positional(sqlite3 *db, double *per_chunk, resume_point *points, + int *chunks_out, long long *bytes_out, int64_t *until_out) { const char *first_sql = "SELECT payload, next_db_version, next_seq, next_frag_offset, is_final, watermark_db_version " "FROM cloudsync_payload_chunks WHERE since_db_version=0 LIMIT 1;"; @@ -68,10 +131,12 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) int chunks = 0; long long bytes = 0; - long long watermark = 0, rdbv = 0, rseq = 0, rfrag = 0; + int64_t watermark = 0, rdbv = 0, rseq = 0, rfrag = 0; bool is_final = false; + double t0 = monotonic_ms(); rc = sqlite3_step(first); + double dt = monotonic_ms() - t0; if (rc == SQLITE_ROW) { bytes += sqlite3_column_bytes(first, 0); rdbv = sqlite3_column_int64(first, 1); @@ -79,6 +144,7 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) rfrag = sqlite3_column_int64(first, 3); is_final = sqlite3_column_int(first, 4) != 0; watermark = sqlite3_column_int64(first, 5); + if (per_chunk) per_chunk[0] = dt; chunks++; } else if (rc == SQLITE_DONE) { rc = SQLITE_OK; @@ -88,23 +154,30 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) } while (!is_final) { + if (points && chunks < MAX_POINTS) { + points[chunks].dbv = rdbv; points[chunks].seq = rseq; points[chunks].frag = rfrag; + } sqlite3_reset(resume); sqlite3_bind_int64(resume, 1, watermark); sqlite3_bind_int64(resume, 2, rdbv); sqlite3_bind_int64(resume, 3, rseq); sqlite3_bind_int64(resume, 4, rfrag); + t0 = monotonic_ms(); rc = sqlite3_step(resume); + dt = monotonic_ms() - t0; if (rc != SQLITE_ROW) { if (rc == SQLITE_DONE) rc = SQLITE_OK; break; } bytes += sqlite3_column_bytes(resume, 0); rdbv = sqlite3_column_int64(resume, 1); rseq = sqlite3_column_int64(resume, 2); rfrag = sqlite3_column_int64(resume, 3); is_final = sqlite3_column_int(resume, 4) != 0; + if (per_chunk && chunks < MAX_POINTS) per_chunk[chunks] = dt; chunks++; } rc = SQLITE_OK; *chunks_out = chunks; *bytes_out = bytes; + *until_out = watermark; done: if (first) sqlite3_finalize(first); @@ -112,11 +185,67 @@ static int drain_positional(sqlite3 *db, int *chunks_out, long long *bytes_out) return rc; } +// Mean of the samples falling in decile b, so a curve over hundreds of chunks +// reads as ten numbers. Returns -1 for an empty bucket. +static double bucket_mean(const double *v, int n, int b) { + double sum = 0; int count = 0; + for (int i = 0; i < n; ++i) { + if (i * BUCKETS / n != b) continue; + sum += v[i]; count++; + } + return count ? sum / count : -1.0; +} + +static void report_curve(const char *title, const double *v, int n) { + printf("%s\n decile:", title); + for (int b = 0; b < BUCKETS; ++b) printf(" %7d%%", (b + 1) * 10); + printf("\n ms/call:"); + for (int b = 0; b < BUCKETS; ++b) { + double m = bucket_mean(v, n, b); + if (m < 0) printf(" --"); else printf(" %8.3f", m); + } + double head = bucket_mean(v, n, 0), tail = bucket_mean(v, n, BUCKETS - 1); + printf("\n last/first decile: "); + if (head > 0 && tail > 0) printf("%.2fx\n", tail / head); else printf("n/a\n"); +} + +// Time one shape at every recorded resume point. Returns the first row each query +// produced so the caller can prove the two shapes select identically. +static int probe_shape(sqlite3 *db, const char *sql, int64_t until, const void *site_id, int site_len, + const resume_point *points, int n, double *out, resume_point *rows) { + sqlite3_stmt *stmt = NULL; + int rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL); + if (rc != SQLITE_OK) { fprintf(stderr, "prepare failed: %s\n", sqlite3_errmsg(db)); return rc; } + for (int i = 0; i < n; ++i) { + sqlite3_reset(stmt); + sqlite3_bind_int64(stmt, 1, until); + sqlite3_bind_blob(stmt, 2, site_id, site_len, SQLITE_TRANSIENT); + sqlite3_bind_int64(stmt, 3, points[i].dbv); + sqlite3_bind_int64(stmt, 4, points[i].dbv); + sqlite3_bind_int64(stmt, 5, points[i].seq); + sqlite3_bind_int64(stmt, 6, points[i].dbv); // SQLITE_RANGE where unused + double t0 = monotonic_ms(); + rc = sqlite3_step(stmt); + out[i] = monotonic_ms() - t0; + if (rc == SQLITE_ROW) { + rows[i].dbv = sqlite3_column_int64(stmt, 0); + rows[i].seq = sqlite3_column_int64(stmt, 1); + } else { + rows[i].dbv = -1; rows[i].seq = -1; + } + } + sqlite3_finalize(stmt); + return SQLITE_OK; +} + int main(void) { - int rows = env_int("CHUNK_BENCH_ROWS", 400); - int row_bytes = env_int("CHUNK_BENCH_ROW_BYTES", 60000); - int repeats = env_int("CHUNK_BENCH_REPEATS", 5); + int rows = env_int("CHUNK_BENCH_ROWS", 3000); + int row_bytes = env_int("CHUNK_BENCH_ROW_BYTES", 8000); + int repeats = env_int("CHUNK_BENCH_REPEATS", 3); int chunk_size = env_int("CHUNK_BENCH_CHUNK_SIZE", 262144); + int txns = env_int("CHUNK_BENCH_TXNS", rows); + bool verbose = env_int("CHUNK_BENCH_VERBOSE", 0) == 1; + if (txns > rows) txns = rows; remove(DB_PATH); sqlite3 *db = NULL; @@ -131,13 +260,11 @@ int main(void) { "SELECT cloudsync_set('payload_max_chunk_size', '%d');", chunk_size); if (db_exec(db, setup) != SQLITE_OK) return 1; - // Split the rows across CHUNK_BENCH_TXNS transactions: each is one db_version, - // so TXNS=1 is the pathological single-version window and TXNS=rows is the - // many-versions case a real /check window resembles. Incompressible bodies keep - // the window many-chunked. - int txns = env_int("CHUNK_BENCH_TXNS", 1); - if (txns < 1) txns = 1; - if (txns > rows) txns = rows; + // Each transaction is one db_version, so TXNS controls how many distinct + // db_versions the window spans. That is the axis the proposed lower bound acts + // on: with TXNS=1 every row shares one db_version and no bound on db_version can + // narrow anything. Incompressible bodies keep the window many-chunked. + printf("seeding %d rows of %d bytes across %d transaction(s)...\n", rows, row_bytes, txns); int idbase = 0; for (int t = 0; t < txns; ++t) { int n = rows / txns + (t < rows % txns ? 1 : 0); @@ -151,26 +278,93 @@ int main(void) { idbase += n; } + double *per_chunk = calloc(MAX_POINTS, sizeof(double)); + resume_point *points = calloc(MAX_POINTS, sizeof(resume_point)); + if (!per_chunk || !points) { fprintf(stderr, "oom\n"); return 1; } + + // ---- Phase 1: the real drain, one chunk per call ---- int chunks = 0; long long bytes = 0; + int64_t until = 0; double best = 1e18, sum = 0; for (int r = 0; r < repeats; ++r) { double t0 = monotonic_ms(); - if (drain_positional(db, &chunks, &bytes) != SQLITE_OK) { fprintf(stderr, "positional drain failed\n"); return 1; } + if (drain_positional(db, per_chunk, points, &chunks, &bytes, &until) != SQLITE_OK) { + fprintf(stderr, "positional drain failed: %s\n", sqlite3_errmsg(db)); + return 1; + } double dt = monotonic_ms() - t0; sum += dt; if (dt < best) best = dt; } + int n = chunks < MAX_POINTS ? chunks : MAX_POINTS; - double mean = sum / repeats; - printf("\nPositional /check drain benchmark (local SQLite, no network)\n"); + printf("\n=== Positional /check drain (local SQLite, no network) ===\n"); printf("rows: %d row_bytes: %d txns: %d chunk_size: %d repeats: %d\n", rows, row_bytes, txns, chunk_size, repeats); - printf("chunks: %d payload_bytes: %lld\n", chunks, bytes); - printf("drain: best=%.2f ms mean=%.2f ms\n", best, mean); - if (chunks > 0) - printf("per-chunk: best=%.3f ms throughput: %.1f MB/s\n", - best / chunks, (double)bytes / 1024.0 / 1024.0 / (best / 1000.0)); + printf("chunks: %d payload_bytes: %lld until_db_version: %lld\n", chunks, bytes, (long long)until); + printf("drain: best=%.2f ms mean=%.2f ms per-chunk=%.3f ms\n", + best, sum / repeats, chunks ? best / chunks : 0.0); + if (n > 0) report_curve("\nPhase 1 - cloudsync_payload_chunks latency vs chunk index", per_chunk, n); + if (verbose) for (int i = 0; i < n; ++i) printf(" chunk %5d %8.3f ms\n", i, per_chunk[i]); + + // ---- Phase 2: the same resume points, straight at cloudsync_changes ---- + // points[0] is unset (the first chunk carries no resume point), so start at 1. + int np = n > 1 ? n - 1 : 0; + if (np > 0) { + // Exclude a site_id that cannot exist, so the filter matches the real + // query's shape without removing any row. + static const unsigned char absent_site[16] = {0}; + static const char *shape_sql[3] = { SHAPE_CURRENT, SHAPE_PROPOSED, SHAPE_REUSED_PARAM }; + static const char *shape_name[3] = { + "2a - current (disjunction, distinct parameters)", + "2b - proposed (+ redundant AND db_version>=?)", + "2c - variant (one parameter reused in both arms)", + }; + static const char *shape_tag[3] = { "current:", "proposed:", "reused:" }; + double *t[3]; + resume_point *sel[3]; + for (int s = 0; s < 3; ++s) { + t[s] = calloc(np, sizeof(double)); + sel[s] = calloc(np, sizeof(resume_point)); + if (!t[s] || !sel[s]) { fprintf(stderr, "oom\n"); return 1; } + } + + printf("\n=== Resume seek against cloudsync_changes, %d points ===\n", np); + printf("clause reaching xBestIndex (a missing db_version >= ? is the replay):\n"); + for (int s = 0; s < 3; ++s) print_plan(db, shape_tag[s], shape_sql[s]); + + for (int s = 0; s < 3; ++s) + if (probe_shape(db, shape_sql[s], until, absent_site, (int)sizeof(absent_site), + points + 1, np, t[s], sel[s]) != SQLITE_OK) return 1; + + for (int s = 0; s < 3; ++s) { + char title[128]; + snprintf(title, sizeof(title), "\nPhase %s", shape_name[s]); + report_curve(title, t[s], np); + } + + printf("\ntotal seek time:"); + for (int s = 0; s < 3; ++s) { + double total = 0; + for (int i = 0; i < np; ++i) total += t[s][i]; + printf(" %s %.2f ms", shape_tag[s], total); + } + printf("\n"); + + // The point of the redundant term is that it is logically implied, so every + // shape must select the identical row at every resume point. + for (int s = 1; s < 3; ++s) { + int mismatches = 0; + for (int i = 0; i < np; ++i) + if (sel[0][i].dbv != sel[s][i].dbv || sel[0][i].seq != sel[s][i].seq) mismatches++; + printf("rows selected by %-9s differ from current at %d of %d resume points\n", + shape_tag[s], mismatches, np); + } + + for (int s = 0; s < 3; ++s) { free(t[s]); free(sel[s]); } + } + free(per_chunk); free(points); sqlite3_close(db); remove(DB_PATH); return 0; From 37e93de8281869ecf7dd6737d8a0a5e600f2d209 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Wed, 23 Sep 2026 08:22:46 -0600 Subject: [PATCH 3/4] test: let chunk-bench measure a production-sized window A window the size of the one in the stall incident (~1 GB, 5 MiB chunks) costs 21s to seed and minutes to drain unfixed, so seeding it per run and per build is not workable. Add CHUNK_BENCH_DB/KEEP/REUSE so one seeded database can be measured by two builds of the extension in turn, CHUNK_BENCH_EXT to pick the build, and CHUNK_BENCH_PHASE2=0 to skip Phase 2, which is itself quadratic and dominates at that size. Flags now parse through env_flag. env_int treats 0 as unset, so that a stray empty value cannot ask for zero rows, which silently made CHUNK_BENCH_PHASE2=0 a no-op. Co-Authored-By: Claude Opus 5 (1M context) --- test/chunk_bench.c | 90 ++++++++++++++++++++++++++++++++++------------ 1 file changed, 68 insertions(+), 22 deletions(-) diff --git a/test/chunk_bench.c b/test/chunk_bench.c index 010b6b52..003e2cdc 100644 --- a/test/chunk_bench.c +++ b/test/chunk_bench.c @@ -26,6 +26,12 @@ // CHUNK_BENCH_REPEATS (default 3), CHUNK_BENCH_CHUNK_SIZE (default 262144), // CHUNK_BENCH_VERBOSE (1 = print every chunk, not just deciles). // +// For a production-sized window, seed once and measure two builds against it: +// CHUNK_BENCH_DB (database path), CHUNK_BENCH_KEEP=1 (do not delete it), +// CHUNK_BENCH_REUSE=1 (skip seeding when it already exists), +// CHUNK_BENCH_EXT (which cloudsync to load), CHUNK_BENCH_PHASE2=0 (skip +// Phase 2, which is itself quadratic and dominates at that size). +// #include #include @@ -40,6 +46,18 @@ #define MAX_POINTS 20000 #define BUCKETS 10 +static const char *env_str(const char *name, const char *dflt) { + const char *v = getenv(name); + return (v && *v) ? v : dflt; +} + +static bool file_exists(const char *path) { + FILE *f = fopen(path, "rb"); + if (!f) return false; + fclose(f); + return true; +} + // Every shape binds ?1=until, ?2=site_id, ?3/?4/?6=resume_db_version, ?5=resume_seq, // and selects exactly the same rows. Only the spelling of the lower bound differs. @@ -86,6 +104,14 @@ static int env_int(const char *name, int dflt) { return (int)p; } +// env_int treats 0 as "unset" so that a stray empty value cannot ask for zero rows. +// Flags need the opposite, since 0 is how you turn one off. +static bool env_flag(const char *name, bool dflt) { + const char *v = getenv(name); + if (!v || !*v) return dflt; + return !(strcmp(v, "0") == 0 || strcmp(v, "false") == 0 || strcmp(v, "no") == 0); +} + static int db_exec(sqlite3 *db, const char *sql) { char *err = NULL; int rc = sqlite3_exec(db, sql, NULL, NULL, &err); @@ -244,38 +270,58 @@ int main(void) { int repeats = env_int("CHUNK_BENCH_REPEATS", 3); int chunk_size = env_int("CHUNK_BENCH_CHUNK_SIZE", 262144); int txns = env_int("CHUNK_BENCH_TXNS", rows); - bool verbose = env_int("CHUNK_BENCH_VERBOSE", 0) == 1; + bool verbose = env_flag("CHUNK_BENCH_VERBOSE", false); + // A large window costs minutes to seed and the unfixed arm re-reads it on every + // call, so allow one seeded database to be reused across both builds and the + // quadratic Phase 2 to be skipped. + const char *db_path = env_str("CHUNK_BENCH_DB", DB_PATH); + const char *ext_path = env_str("CHUNK_BENCH_EXT", EXT_PATH); + bool keep = env_flag("CHUNK_BENCH_KEEP", false); + bool reuse = env_flag("CHUNK_BENCH_REUSE", false) && file_exists(db_path); + bool want_phase2 = env_flag("CHUNK_BENCH_PHASE2", true); if (txns > rows) txns = rows; - remove(DB_PATH); + if (!reuse) remove(db_path); sqlite3 *db = NULL; - if (sqlite3_open(DB_PATH, &db) != SQLITE_OK) { fprintf(stderr, "open failed\n"); return 1; } + if (sqlite3_open(db_path, &db) != SQLITE_OK) { fprintf(stderr, "open failed\n"); return 1; } if (sqlite3_enable_load_extension(db, 1) != SQLITE_OK) return 1; - if (db_exec(db, "SELECT load_extension('" EXT_PATH "');") != SQLITE_OK) return 1; + char load[512]; + snprintf(load, sizeof(load), "SELECT load_extension('%s');", ext_path); + if (db_exec(db, load) != SQLITE_OK) return 1; + printf("extension: %s database: %s%s\n", ext_path, db_path, reuse ? " (reused)" : ""); char setup[256]; - snprintf(setup, sizeof(setup), - "CREATE TABLE chunk_bench (id TEXT PRIMARY KEY, body BLOB);" - "SELECT cloudsync_init('chunk_bench');" - "SELECT cloudsync_set('payload_max_chunk_size', '%d');", chunk_size); + snprintf(setup, sizeof(setup), "SELECT cloudsync_set('payload_max_chunk_size', '%d');", chunk_size); + if (!reuse) { + char create[256]; + snprintf(create, sizeof(create), + "PRAGMA journal_mode=WAL; PRAGMA synchronous=OFF;" + "CREATE TABLE chunk_bench (id TEXT PRIMARY KEY, body BLOB);" + "SELECT cloudsync_init('chunk_bench');"); + if (db_exec(db, create) != SQLITE_OK) return 1; + } if (db_exec(db, setup) != SQLITE_OK) return 1; // Each transaction is one db_version, so TXNS controls how many distinct // db_versions the window spans. That is the axis the proposed lower bound acts // on: with TXNS=1 every row shares one db_version and no bound on db_version can // narrow anything. Incompressible bodies keep the window many-chunked. - printf("seeding %d rows of %d bytes across %d transaction(s)...\n", rows, row_bytes, txns); - int idbase = 0; - for (int t = 0; t < txns; ++t) { - int n = rows / txns + (t < rows % txns ? 1 : 0); - if (n <= 0) continue; - char insert[256]; - snprintf(insert, sizeof(insert), - "WITH RECURSIVE c(i) AS (SELECT %d UNION ALL SELECT i+1 FROM c WHERE i < %d) " - "INSERT INTO chunk_bench(id, body) SELECT printf('row-%%06d', i), randomblob(%d) FROM c;", - idbase + 1, idbase + n, row_bytes); - if (db_exec(db, insert) != SQLITE_OK) return 1; - idbase += n; + if (!reuse) { + printf("seeding %d rows of %d bytes across %d transaction(s)...\n", rows, row_bytes, txns); + double seed_t0 = monotonic_ms(); + int idbase = 0; + for (int t = 0; t < txns; ++t) { + int n = rows / txns + (t < rows % txns ? 1 : 0); + if (n <= 0) continue; + char insert[256]; + snprintf(insert, sizeof(insert), + "WITH RECURSIVE c(i) AS (SELECT %d UNION ALL SELECT i+1 FROM c WHERE i < %d) " + "INSERT INTO chunk_bench(id, body) SELECT printf('row-%%06d', i), randomblob(%d) FROM c;", + idbase + 1, idbase + n, row_bytes); + if (db_exec(db, insert) != SQLITE_OK) return 1; + idbase += n; + } + printf("seeded in %.1f s\n", (monotonic_ms() - seed_t0) / 1000.0); } double *per_chunk = calloc(MAX_POINTS, sizeof(double)); @@ -309,7 +355,7 @@ int main(void) { // ---- Phase 2: the same resume points, straight at cloudsync_changes ---- // points[0] is unset (the first chunk carries no resume point), so start at 1. - int np = n > 1 ? n - 1 : 0; + int np = (want_phase2 && n > 1) ? n - 1 : 0; if (np > 0) { // Exclude a site_id that cannot exist, so the filter matches the real // query's shape without removing any row. @@ -366,6 +412,6 @@ int main(void) { free(per_chunk); free(points); sqlite3_close(db); - remove(DB_PATH); + if (!keep) remove(db_path); return 0; } From 92bcce9a9b70a06bc3487cc1bdc523808ae77835 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Wed, 23 Sep 2026 09:22:33 -0600 Subject: [PATCH 4/4] fix(test): report a failed resume instead of counting it as a short drain Review findings on this PR: drain_positional overwrote the step result with SQLITE_OK after the loop, so a failed resume returned success with fewer chunks -- and a truncated drain is indistinguishable from a fast one, which is the exact number this benchmark exists to produce. Report it and propagate. SQLITE_DONE counts as a failure too: the loop only runs while the previous chunk said it was not final, so the stream still owes a chunk. Pre-existing, from when the file was added. The benchmark called the unfixed spelling "current" and the fixed one "proposed", which inverts once this PR lands. They are now "old" and "fixed", so Phase 2 reads as the regression check it becomes after merge. Drop the pointer to docs/internal/payload-chunks-resume-scan.md from the comment in payload_chunks_filter(): that note is not committed, so a reader of the public source cannot follow it. The comment now carries the reasoning. Cite the function rather than line numbers, which had already gone stale. Co-Authored-By: Claude Opus 5 (1M context) --- src/sqlite/cloudsync_sqlite.c | 10 ++++----- test/chunk_bench.c | 42 +++++++++++++++++++++-------------- 2 files changed, 30 insertions(+), 22 deletions(-) diff --git a/src/sqlite/cloudsync_sqlite.c b/src/sqlite/cloudsync_sqlite.c index a64ab1c8..8345dd81 100644 --- a/src/sqlite/cloudsync_sqlite.c +++ b/src/sqlite/cloudsync_sqlite.c @@ -1401,11 +1401,11 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const // (db_version, seq) >= (resume_dbv, resume_seq). char *sql; if (positional) { - // The redundant db_version>=? is what makes the resume a seek. The - // disjunction alone states the same bound, but its two arms carry distinct - // parameters, so SQLite cannot derive a range from it and cloudsync_changes' - // xBestIndex is offered no lower bound at all — leaving every call to replay - // the window from the start. See docs/internal/payload-chunks-resume-scan.md. + // The redundant db_version>=? is what makes the resume a seek, and it is + // load-bearing: SQLite derives a range from a disjunction only when both arms + // compare against the same value, and these two arms carry distinct + // parameters. Without it cloudsync_changes' xBestIndex is offered no lower + // bound at all and every call replays the window from the start. sql = sqlite3_mprintf( "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq " "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND db_version>=? AND " diff --git a/test/chunk_bench.c b/test/chunk_bench.c index 003e2cdc..0bc37a12 100644 --- a/test/chunk_bench.c +++ b/test/chunk_bench.c @@ -12,10 +12,10 @@ // the window, making the drain O(N^2); flat latency means the resume seeks. // // Phase 2 — replay the same resume points straight against cloudsync_changes -// in two SQL shapes: the one the positional branch emits today, and the same -// clause plus the redundant AND-connected `db_version>=?` lower bound -// proposed in docs/internal/payload-chunks-resume-scan.md. Same rows, two -// plans, so the difference is the value of that proposal. +// in three SQL shapes: the lower bound as payload_chunks_filter() spelled it +// before the fix, the shape it uses now, and a variant. Same rows, different +// plans, so the difference is attributable to the spelling alone. Keeping the +// old shape here is what makes this a regression check rather than a one-off. // // CI builds this with the other test binaries but never runs it: the timings are // machine-dependent. Run it by hand with `make chunk-bench`. @@ -61,17 +61,17 @@ static bool file_exists(const char *path) { // Every shape binds ?1=until, ?2=site_id, ?3/?4/?6=resume_db_version, ?5=resume_seq, // and selects exactly the same rows. Only the spelling of the lower bound differs. -// What the positional branch emits today (cloudsync_sqlite.c:1403-1414): the resume -// point lives entirely inside a disjunction whose two arms carry *distinct* -// parameters. Distinct parameters are what defeats the planner here — see below. -#define SHAPE_CURRENT \ +// What payload_chunks_filter() emitted before the fix: the resume point lived +// entirely inside a disjunction whose two arms carry *distinct* parameters. +// Distinct parameters are what defeats the planner here — see below. +#define SHAPE_OLD \ "SELECT db_version, seq FROM cloudsync_changes " \ "WHERE db_version<=?1 AND site_id<>?2 AND (db_version>?3 OR (db_version=?4 AND seq>=?5)) " \ "ORDER BY db_version, seq ASC LIMIT 1" -// The doc's proposal: a redundant conjunct stating the lower bound where a virtual +// What it emits now: a redundant conjunct stating the lower bound where a virtual // table's xBestIndex can see it, alongside the untouched disjunction. -#define SHAPE_PROPOSED \ +#define SHAPE_FIXED \ "SELECT db_version, seq FROM cloudsync_changes " \ "WHERE db_version<=?1 AND site_id<>?2 AND db_version>=?6 AND (db_version>?3 OR (db_version=?4 AND seq>=?5)) " \ "ORDER BY db_version, seq ASC LIMIT 1" @@ -191,7 +191,15 @@ static int drain_positional(sqlite3 *db, double *per_chunk, resume_point *points t0 = monotonic_ms(); rc = sqlite3_step(resume); dt = monotonic_ms() - t0; - if (rc != SQLITE_ROW) { if (rc == SQLITE_DONE) rc = SQLITE_OK; break; } + // A drain that stops early looks exactly like a fast one, so never report it + // as success. SQLITE_DONE here is a failure too: the loop only runs while the + // previous chunk said it was not final, so the stream owes us another chunk. + if (rc != SQLITE_ROW) { + fprintf(stderr, "resume failed at chunk %d: %s\n", chunks, + (rc == SQLITE_DONE) ? "no chunk returned before is_final" : sqlite3_errmsg(db)); + if (rc == SQLITE_DONE) rc = SQLITE_ERROR; + goto done; + } bytes += sqlite3_column_bytes(resume, 0); rdbv = sqlite3_column_int64(resume, 1); rseq = sqlite3_column_int64(resume, 2); @@ -360,13 +368,13 @@ int main(void) { // Exclude a site_id that cannot exist, so the filter matches the real // query's shape without removing any row. static const unsigned char absent_site[16] = {0}; - static const char *shape_sql[3] = { SHAPE_CURRENT, SHAPE_PROPOSED, SHAPE_REUSED_PARAM }; + static const char *shape_sql[3] = { SHAPE_OLD, SHAPE_FIXED, SHAPE_REUSED_PARAM }; static const char *shape_name[3] = { - "2a - current (disjunction, distinct parameters)", - "2b - proposed (+ redundant AND db_version>=?)", - "2c - variant (one parameter reused in both arms)", + "2a - old (disjunction, distinct parameters)", + "2b - fixed (+ redundant AND db_version>=?)", + "2c - variant (one parameter reused in both arms)", }; - static const char *shape_tag[3] = { "current:", "proposed:", "reused:" }; + static const char *shape_tag[3] = { "old:", "fixed:", "reused:" }; double *t[3]; resume_point *sel[3]; for (int s = 0; s < 3; ++s) { @@ -403,7 +411,7 @@ int main(void) { int mismatches = 0; for (int i = 0; i < np; ++i) if (sel[0][i].dbv != sel[s][i].dbv || sel[0][i].seq != sel[s][i].seq) mismatches++; - printf("rows selected by %-9s differ from current at %d of %d resume points\n", + printf("rows selected by %-7s differ from the old shape at %d of %d resume points\n", shape_tag[s], mismatches, np); }