refactor(payload): remove the cloudsync_payload_spool table and functions

The /check job now pages the tenant via the positional cursor on
cloudsync_payload_chunks, so the server-side download spool is dead.
Remove the cloudsync_payload_spool table and the
cloudsync_payload_spool_fill / _drop / _drop_chunk functions on both
engines (SQLite: sql_sqlite.c constants, sql.h decls, cloudsync_sqlite.c
functions + registration; PostgreSQL: cloudsync.sql.in and the 1.0->1.1
migration), plus the SQLite "Payload Download Spool" unit test and the
spool sections of the PostgreSQL chunk test.

The client-side cursor paging (network.c) is unchanged: the wire protocol
still pages chunks by an integer cursor; only the server's node-side
staging mechanism changed. All remaining chunk/fragment/positional tests
pass on both engines.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: andinux

src/postgresql/cloudsync.sql.in

@@ -198,119 +198,6 @@ RETURNS bytea
 AS 'MODULE_PATHNAME', 'cloudsync_uuid_blob'
 LANGUAGE C IMMUTABLE;
 
--- Download spool (server-side /check chunk staging)
---
--- The /check endpoint runs on a separate host and reaches the tenant DB over a
--- network driver, which materializes the whole result set. Streaming chunks
--- directly with SELECT * FROM cloudsync_payload_chunks(...) would therefore pull
--- every chunk into the server's memory at once. Instead the server fills a
--- window's chunk stream once into this table and pages it out one chunk per call.
---
--- UNLOGGED: this is regenerable scratch state, so skip WAL. Created at install
--- time (not lazily inside the function) to avoid the plpgsql create-then-use
--- cached-plan pitfall.
-CREATE TABLE IF NOT EXISTS cloudsync_payload_spool (
-  stream_id text NOT NULL,
-  chunk_index bigint NOT NULL,
-  payload bytea NOT NULL,
-  payload_size bigint NOT NULL,
-  db_version_min bigint NOT NULL,
-  db_version_max bigint NOT NULL,
-  watermark bigint NOT NULL,
-  is_final boolean NOT NULL DEFAULT false,
-  created_at bigint NOT NULL DEFAULT extract(epoch FROM now())::bigint,
-  PRIMARY KEY (stream_id, chunk_index)
-);
-
--- cloudsync_payload_spool_fill(stream_id, since, filter_site_id, exclude)
--- Generate the whole chunk stream for a window once into cloudsync_payload_spool.
--- Returns the number of chunks spooled. Idempotent: a prior complete fill is kept.
--- Parameters are p_-prefixed so they never collide with the table's columns
--- (an unqualified `stream_id` would otherwise be ambiguous with the parameter).
-CREATE OR REPLACE FUNCTION cloudsync_payload_spool_fill(
-  p_stream_id text,
-  p_since_db_version bigint,
-  p_filter_site_id bytea DEFAULT NULL,
-  p_exclude_filter_site_id boolean DEFAULT false
-) RETURNS bigint AS $$
-DECLARE
-  existing bigint;
-  cnt bigint := 0;
-  rec record;
-BEGIN
-  -- Stale-GC of abandoned streams. fill runs once per stream (coarse-grained),
-  -- so unlike per-fragment cleanup there is no O(n^2) risk and no throttle.
-  DELETE FROM cloudsync_payload_spool
-  WHERE stream_id IN (
-    SELECT s.stream_id FROM cloudsync_payload_spool s
-    GROUP BY s.stream_id
-    HAVING max(s.created_at) < extract(epoch FROM now())::bigint - 86400);
-
-  -- Idempotent: a prior complete fill stays as-is (fill is atomic, so rows
-  -- present == complete stream).
-  SELECT count(*) INTO existing FROM cloudsync_payload_spool WHERE stream_id = p_stream_id;
-  IF existing > 0 THEN
-    RETURN existing;
-  END IF;
-
-  -- Generate the stream one chunk at a time. A cursor FOR loop fetches in
-  -- batches rather than materializing the whole SRF into a tuplestore.
-  FOR rec IN
-    SELECT c.payload, c.chunk_index, c.payload_size, c.db_version_min, c.db_version_max, c.watermark_db_version
-    FROM cloudsync_payload_chunks(p_since_db_version, p_filter_site_id, NULL, p_exclude_filter_site_id) c
-  LOOP
-    INSERT INTO cloudsync_payload_spool
-      (stream_id, chunk_index, payload, payload_size, db_version_min, db_version_max, watermark, is_final)
-    VALUES (p_stream_id, rec.chunk_index, rec.payload, rec.payload_size,
-            rec.db_version_min, rec.db_version_max, rec.watermark_db_version, false);
-    cnt := cnt + 1;
-  END LOOP;
-
-  IF cnt > 0 THEN
-    UPDATE cloudsync_payload_spool SET is_final = true
-    WHERE stream_id = p_stream_id
-      AND chunk_index = (SELECT max(x.chunk_index) FROM cloudsync_payload_spool x
-                         WHERE x.stream_id = p_stream_id);
-  END IF;
-
-  RETURN cnt;
-END;
-$$ LANGUAGE plpgsql VOLATILE;
-
--- cloudsync_payload_spool_drop(stream_id) -> number of chunks removed.
-CREATE OR REPLACE FUNCTION cloudsync_payload_spool_drop(p_stream_id text)
-RETURNS bigint AS $$
-DECLARE
-  deleted bigint := 0;
-BEGIN
-  DELETE FROM cloudsync_payload_spool WHERE stream_id = p_stream_id;
-  GET DIAGNOSTICS deleted = ROW_COUNT;
-  RETURN deleted;
-END;
-$$ LANGUAGE plpgsql VOLATILE;
-
--- cloudsync_payload_spool_drop_chunk(stream_id, chunk_index) -> number of chunks removed.
--- Explicit early cleanup for one S3-backed chunk; stream-level TTL/GC remains unchanged.
-CREATE OR REPLACE FUNCTION cloudsync_payload_spool_drop_chunk(p_stream_id text, p_chunk_index bigint)
-RETURNS bigint AS $$
-DECLARE
-  deleted bigint := 0;
-BEGIN
-  IF p_stream_id IS NULL THEN
-    RAISE EXCEPTION 'cloudsync_payload_spool_drop_chunk: stream_id is required.';
-  END IF;
-  IF p_chunk_index IS NULL THEN
-    RAISE EXCEPTION 'cloudsync_payload_spool_drop_chunk: chunk_index is required.';
-  END IF;
-
-  DELETE FROM cloudsync_payload_spool
-  WHERE stream_id = p_stream_id
-    AND chunk_index = p_chunk_index;
-  GET DIAGNOSTICS deleted = ROW_COUNT;
-  RETURN deleted;
-END;
-$$ LANGUAGE plpgsql VOLATILE;
-
 -- Payload decoding and application
 CREATE OR REPLACE FUNCTION cloudsync_payload_decode(payload bytea)
 RETURNS integer

src/postgresql/migrations/cloudsync--1.0--1.1.sql

@@ -55,93 +55,3 @@ RETURNS bytea
 AS 'MODULE_PATHNAME', 'cloudsync_uuid_blob'
 LANGUAGE C IMMUTABLE;
 
--- Download spool: the /check path fills a window's chunk stream once and pages it
--- out one chunk per call so the network driver never re-materializes the whole
--- stream. See cloudsync.sql.in for the rationale.
-CREATE TABLE IF NOT EXISTS cloudsync_payload_spool (
-  stream_id text NOT NULL,
-  chunk_index bigint NOT NULL,
-  payload bytea NOT NULL,
-  payload_size bigint NOT NULL,
-  db_version_min bigint NOT NULL,
-  db_version_max bigint NOT NULL,
-  watermark bigint NOT NULL,
-  is_final boolean NOT NULL DEFAULT false,
-  created_at bigint NOT NULL DEFAULT extract(epoch FROM now())::bigint,
-  PRIMARY KEY (stream_id, chunk_index)
-);
-
-CREATE OR REPLACE FUNCTION cloudsync_payload_spool_fill(
-  p_stream_id text,
-  p_since_db_version bigint,
-  p_filter_site_id bytea DEFAULT NULL,
-  p_exclude_filter_site_id boolean DEFAULT false
-) RETURNS bigint AS $$
-DECLARE
-  existing bigint;
-  cnt bigint := 0;
-  rec record;
-BEGIN
-  DELETE FROM cloudsync_payload_spool
-  WHERE stream_id IN (
-    SELECT s.stream_id FROM cloudsync_payload_spool s
-    GROUP BY s.stream_id
-    HAVING max(s.created_at) < extract(epoch FROM now())::bigint - 86400);
-
-  SELECT count(*) INTO existing FROM cloudsync_payload_spool WHERE stream_id = p_stream_id;
-  IF existing > 0 THEN
-    RETURN existing;
-  END IF;
-
-  FOR rec IN
-    SELECT c.payload, c.chunk_index, c.payload_size, c.db_version_min, c.db_version_max, c.watermark_db_version
-    FROM cloudsync_payload_chunks(p_since_db_version, p_filter_site_id, NULL, p_exclude_filter_site_id) c
-  LOOP
-    INSERT INTO cloudsync_payload_spool
-      (stream_id, chunk_index, payload, payload_size, db_version_min, db_version_max, watermark, is_final)
-    VALUES (p_stream_id, rec.chunk_index, rec.payload, rec.payload_size,
-            rec.db_version_min, rec.db_version_max, rec.watermark_db_version, false);
-    cnt := cnt + 1;
-  END LOOP;
-
-  IF cnt > 0 THEN
-    UPDATE cloudsync_payload_spool SET is_final = true
-    WHERE stream_id = p_stream_id
-      AND chunk_index = (SELECT max(x.chunk_index) FROM cloudsync_payload_spool x
-                         WHERE x.stream_id = p_stream_id);
-  END IF;
-
-  RETURN cnt;
-END;
-$$ LANGUAGE plpgsql VOLATILE;
-
-CREATE OR REPLACE FUNCTION cloudsync_payload_spool_drop(p_stream_id text)
-RETURNS bigint AS $$
-DECLARE
-  deleted bigint := 0;
-BEGIN
-  DELETE FROM cloudsync_payload_spool WHERE stream_id = p_stream_id;
-  GET DIAGNOSTICS deleted = ROW_COUNT;
-  RETURN deleted;
-END;
-$$ LANGUAGE plpgsql VOLATILE;
-
-CREATE OR REPLACE FUNCTION cloudsync_payload_spool_drop_chunk(p_stream_id text, p_chunk_index bigint)
-RETURNS bigint AS $$
-DECLARE
-  deleted bigint := 0;
-BEGIN
-  IF p_stream_id IS NULL THEN
-    RAISE EXCEPTION 'cloudsync_payload_spool_drop_chunk: stream_id is required.';
-  END IF;
-  IF p_chunk_index IS NULL THEN
-    RAISE EXCEPTION 'cloudsync_payload_spool_drop_chunk: chunk_index is required.';
-  END IF;
-
-  DELETE FROM cloudsync_payload_spool
-  WHERE stream_id = p_stream_id
-    AND chunk_index = p_chunk_index;
-  GET DIAGNOSTICS deleted = ROW_COUNT;
-  RETURN deleted;
-END;
-$$ LANGUAGE plpgsql VOLATILE;

src/sql.h

@@ -74,14 +74,6 @@ extern const char * const SQL_PAYLOAD_FRAGMENTS_SELECT;
 extern const char * const SQL_PAYLOAD_FRAGMENTS_DELETE;
 extern const char * const SQL_PAYLOAD_FRAGMENTS_CLEANUP_STALE;
 
-extern const char * const SQL_PAYLOAD_SPOOL_CREATE_TABLE;
-extern const char * const SQL_PAYLOAD_SPOOL_COUNT;
-extern const char * const SQL_PAYLOAD_SPOOL_FILL_INSERT;
-extern const char * const SQL_PAYLOAD_SPOOL_MARK_FINAL;
-extern const char * const SQL_PAYLOAD_SPOOL_DELETE;
-extern const char * const SQL_PAYLOAD_SPOOL_DELETE_CHUNK;
-extern const char * const SQL_PAYLOAD_SPOOL_CLEANUP_STALE;
-
 // BLOCKS (block-level LWW)
 extern const char * const SQL_BLOCKS_CREATE_TABLE;
 extern const char * const SQL_BLOCKS_UPSERT;

src/sqlite/cloudsync_sqlite.c

@@ -1852,152 +1852,6 @@ void dbsync_payload_load (sqlite3_context *context, int argc, sqlite3_value **ar
 }
 #endif
 
-// MARK: - Download spool -
-
-// Abandoned download streams (a client that started a chunked /check drain and
-// never finished) are reaped after this many seconds. Matches the v3 fragment
-// stale window.
-#define CLOUDSYNC_PAYLOAD_SPOOL_STALE_SECONDS (24*60*60)
-
-// cloudsync_payload_spool_fill(stream_id, since_db_version, filter_site_id, exclude)
-// Generate the whole chunk stream for a window once into cloudsync_payload_spool
-// so the /check path can page it out one chunk per call. Returns the number of
-// chunks spooled for stream_id. Idempotent: a prior complete fill is kept as-is.
-static void dbsync_payload_spool_fill (sqlite3_context *context, int argc, sqlite3_value **argv) {
-    DEBUG_FUNCTION("cloudsync_payload_spool_fill");
-    UNUSED_PARAMETER(argc);
-
-    sqlite3 *db = sqlite3_context_db_handle(context);
-    if (sqlite3_value_type(argv[0]) == SQLITE_NULL) {
-        sqlite3_result_error(context, "cloudsync_payload_spool_fill: stream_id is required.", -1);
-        return;
-    }
-    const char *stream_id = (const char *)sqlite3_value_text(argv[0]);
-    int stream_id_len = sqlite3_value_bytes(argv[0]);
-    int64_t since = sqlite3_value_int64(argv[1]);
-    bool site_given = (sqlite3_value_type(argv[2]) != SQLITE_NULL);
-    int exclude = (sqlite3_value_int(argv[3]) != 0);
-
-    sqlite3_stmt *stmt = NULL;
-    int rc = sqlite3_exec(db, SQL_PAYLOAD_SPOOL_CREATE_TABLE, NULL, NULL, NULL);
-    if (rc != SQLITE_OK) goto error;
-
-    // Stale-GC of abandoned streams (coarse-grained, no throttle needed).
-    if (sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_CLEANUP_STALE, -1, &stmt, NULL) == SQLITE_OK) {
-        sqlite3_bind_int64(stmt, 1, (int64_t)time(NULL) - CLOUDSYNC_PAYLOAD_SPOOL_STALE_SECONDS);
-        sqlite3_step(stmt);
-    }
-    sqlite3_finalize(stmt); stmt = NULL;
-
-    // Atomic fill: a partial/failed generation must never persist, so that the
-    // idempotency check below ("rows present == complete stream") holds.
-    rc = sqlite3_exec(db, "SAVEPOINT cloudsync_spool_fill;", NULL, NULL, NULL);
-    if (rc != SQLITE_OK) goto error;
-
-    int64_t count = 0;
-    rc = sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_COUNT, -1, &stmt, NULL);
-    if (rc != SQLITE_OK) goto rollback;
-    sqlite3_bind_text(stmt, 1, stream_id, stream_id_len, SQLITE_TRANSIENT);
-    if (sqlite3_step(stmt) == SQLITE_ROW) count = sqlite3_column_int64(stmt, 0);
-    sqlite3_finalize(stmt); stmt = NULL;
-
-    if (count == 0) {
-        rc = sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_FILL_INSERT, -1, &stmt, NULL);
-        if (rc != SQLITE_OK) goto rollback;
-        sqlite3_bind_text(stmt, 1, stream_id, stream_id_len, SQLITE_TRANSIENT);
-        sqlite3_bind_int64(stmt, 2, since);
-        if (site_given) sqlite3_bind_blob(stmt, 3, sqlite3_value_blob(argv[2]), sqlite3_value_bytes(argv[2]), SQLITE_TRANSIENT);
-        else sqlite3_bind_null(stmt, 3);
-        sqlite3_bind_int(stmt, 4, exclude);
-        rc = sqlite3_step(stmt);
-        sqlite3_finalize(stmt); stmt = NULL;
-        if (rc != SQLITE_DONE) goto rollback;
-
-        rc = sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_MARK_FINAL, -1, &stmt, NULL);
-        if (rc != SQLITE_OK) goto rollback;
-        sqlite3_bind_text(stmt, 1, stream_id, stream_id_len, SQLITE_TRANSIENT);
-        rc = sqlite3_step(stmt);
-        sqlite3_finalize(stmt); stmt = NULL;
-        if (rc != SQLITE_DONE) goto rollback;
-
-        rc = sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_COUNT, -1, &stmt, NULL);
-        if (rc != SQLITE_OK) goto rollback;
-        sqlite3_bind_text(stmt, 1, stream_id, stream_id_len, SQLITE_TRANSIENT);
-        if (sqlite3_step(stmt) == SQLITE_ROW) count = sqlite3_column_int64(stmt, 0);
-        sqlite3_finalize(stmt); stmt = NULL;
-    }
-
-    sqlite3_exec(db, "RELEASE cloudsync_spool_fill;", NULL, NULL, NULL);
-    sqlite3_result_int64(context, count);
-    return;
-
-rollback:
-    sqlite3_finalize(stmt);
-    sqlite3_result_error(context, sqlite3_errmsg(db), -1);
-    sqlite3_result_error_code(context, rc);
-    sqlite3_exec(db, "ROLLBACK TO cloudsync_spool_fill; RELEASE cloudsync_spool_fill;", NULL, NULL, NULL);
-    return;
-
-error:
-    sqlite3_result_error(context, sqlite3_errmsg(db), -1);
-    sqlite3_result_error_code(context, rc);
-}
-
-// cloudsync_payload_spool_drop(stream_id) -> number of chunks removed.
-// Called once a stream has been fully delivered/acked (or to force-evict).
-static void dbsync_payload_spool_drop (sqlite3_context *context, int argc, sqlite3_value **argv) {
-    DEBUG_FUNCTION("cloudsync_payload_spool_drop");
-    UNUSED_PARAMETER(argc);
-
-    sqlite3 *db = sqlite3_context_db_handle(context);
-    if (sqlite3_value_type(argv[0]) == SQLITE_NULL) {
-        sqlite3_result_error(context, "cloudsync_payload_spool_drop: stream_id is required.", -1);
-        return;
-    }
-
-    int rc = sqlite3_exec(db, SQL_PAYLOAD_SPOOL_CREATE_TABLE, NULL, NULL, NULL);
-    if (rc != SQLITE_OK) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); return; }
-
-    sqlite3_stmt *stmt = NULL;
-    rc = sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_DELETE, -1, &stmt, NULL);
-    if (rc != SQLITE_OK) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); return; }
-    sqlite3_bind_text(stmt, 1, (const char *)sqlite3_value_text(argv[0]), sqlite3_value_bytes(argv[0]), SQLITE_TRANSIENT);
-    rc = sqlite3_step(stmt);
-    sqlite3_finalize(stmt);
-    if (rc != SQLITE_DONE) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); return; }
-    sqlite3_result_int64(context, sqlite3_changes(db));
-}
-
-// cloudsync_payload_spool_drop_chunk(stream_id, chunk_index) -> number of chunks removed.
-// Called after one S3-backed chunk has been safely persisted outside the spool.
-static void dbsync_payload_spool_drop_chunk (sqlite3_context *context, int argc, sqlite3_value **argv) {
-    DEBUG_FUNCTION("cloudsync_payload_spool_drop_chunk");
-    UNUSED_PARAMETER(argc);
-
-    sqlite3 *db = sqlite3_context_db_handle(context);
-    if (sqlite3_value_type(argv[0]) == SQLITE_NULL) {
-        sqlite3_result_error(context, "cloudsync_payload_spool_drop_chunk: stream_id is required.", -1);
-        return;
-    }
-    if (sqlite3_value_type(argv[1]) == SQLITE_NULL) {
-        sqlite3_result_error(context, "cloudsync_payload_spool_drop_chunk: chunk_index is required.", -1);
-        return;
-    }
-
-    int rc = sqlite3_exec(db, SQL_PAYLOAD_SPOOL_CREATE_TABLE, NULL, NULL, NULL);
-    if (rc != SQLITE_OK) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); return; }
-
-    sqlite3_stmt *stmt = NULL;
-    rc = sqlite3_prepare_v2(db, SQL_PAYLOAD_SPOOL_DELETE_CHUNK, -1, &stmt, NULL);
-    if (rc != SQLITE_OK) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); return; }
-    sqlite3_bind_text(stmt, 1, (const char *)sqlite3_value_text(argv[0]), sqlite3_value_bytes(argv[0]), SQLITE_TRANSIENT);
-    sqlite3_bind_int64(stmt, 2, sqlite3_value_int64(argv[1]));
-    rc = sqlite3_step(stmt);
-    sqlite3_finalize(stmt);
-    if (rc != SQLITE_DONE) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); return; }
-    sqlite3_result_int64(context, sqlite3_changes(db));
-}
-
 // MARK: - Register -
 
 int dbsync_register_with_flags (sqlite3 *db, const char *name, void (*xfunc)(sqlite3_context*,int,sqlite3_value**), void (*xstep)(sqlite3_context*,int,sqlite3_value**), void (*xfinal)(sqlite3_context*), int nargs, int flags, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
@@ -2319,14 +2173,6 @@ int dbsync_register_functions (sqlite3 *db, char **pzErrMsg) {
     rc = dbsync_register_function(db, "cloudsync_payload_blob_checked", dbsync_payload_blob_checked, 5, pzErrMsg, ctx, NULL);
     if (rc != SQLITE_OK) return rc;
 
-    // Download spool (server-side /check chunk staging)
-    rc = dbsync_register_function(db, "cloudsync_payload_spool_fill", dbsync_payload_spool_fill, 4, pzErrMsg, ctx, NULL);
-    if (rc != SQLITE_OK) return rc;
-    rc = dbsync_register_function(db, "cloudsync_payload_spool_drop", dbsync_payload_spool_drop, 1, pzErrMsg, ctx, NULL);
-    if (rc != SQLITE_OK) return rc;
-    rc = dbsync_register_function(db, "cloudsync_payload_spool_drop_chunk", dbsync_payload_spool_drop_chunk, 2, pzErrMsg, ctx, NULL);
-    if (rc != SQLITE_OK) return rc;
-
     #ifdef CLOUDSYNC_DESKTOP_OS
     rc = dbsync_register_function(db, "cloudsync_payload_save", dbsync_payload_save, 1, pzErrMsg, ctx, NULL);
     if (rc != SQLITE_OK) return rc;