feat(payload): download spool for chunked /check

Add a server-side spool so the /check download path can page a window's
chunk stream one chunk per call, instead of having the network driver
(libpq / sqlitecloud-go) re-materialize the whole stream into memory.

- cloudsync_payload_spool table + cloudsync_payload_spool_fill/_drop on
  both engines (SQLite C, lazy-create; PG plpgsql, table created at
  install). fill generates a window's chunks once (idempotent, atomic),
  marks the last chunk is_final, and self-GCs abandoned streams (24h TTL).
- cloudsync_network_check_internal echoes a best-effort page cursor
  to/from /check so the stateless server serves the next spool page;
  retrocompatible (optional response field, sent in every request).
- Tests: do_test_payload_spool (SQLite) and a spool block in
  52_payload_chunks.sql (PG) covering byte-identity vs direct generation,
  idempotent re-fill, empty window, drop, and stale-GC.

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

Author: andinux

src/network/network.c

@@ -83,6 +83,13 @@ struct network_data {
     char        *apply_endpoint;
     char        *status_endpoint;
     int         ticket_enabled;
+    // Best-effort page cursor for the chunked /check download drain. The durable
+    // receive cursor (check_dbversion/check_seq) is frozen at "since" for the whole
+    // drain, so the server (which is stateless across /check calls) needs the client
+    // to echo which spool page to serve next. In-memory only: losing it just
+    // restarts the drain from page 0, which is safe because apply is idempotent.
+    int64_t     check_cursor;        // next page index to request (0 = fresh drain)
+    int64_t     check_cursor_since;  // the check_dbversion check_cursor belongs to
 #ifndef CLOUDSYNC_OMIT_CURL
     CURL        *api_curl;
     CURL        *artifact_curl;
@@ -1636,11 +1643,22 @@ int cloudsync_network_check_internal(sqlite3_context *context, int *pnrows, sync
     int seq = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_CHECK_SEQ);
     if (seq<0) {sqlite3_result_error(context, "Unable to retrieve seq.", -1); return -1;}
 
+    // Restart paging whenever the durable receive window changes: the page cursor
+    // is only meaningful within a single drain (check_dbversion held at "since").
+    if (netdata->check_cursor_since != db_version) {
+        netdata->check_cursor = 0;
+        netdata->check_cursor_since = db_version;
+    }
+
     // Capture local db_version before download so we can query cloudsync_changes afterwards
     int64_t prev_dbv = cloudsync_dbversion(data);
 
+    // "cursor" is the spool page to serve. Old/legacy servers ignore the unknown
+    // request field and omit it from the response; the client then never self-pages
+    // (check_cursor stays 0), preserving current behavior.
     char json_payload[2024];
-    snprintf(json_payload, sizeof(json_payload), "{\"dbVersion\":%lld, \"seq\":%d}", (long long)db_version, seq);
+    snprintf(json_payload, sizeof(json_payload), "{\"dbVersion\":%lld, \"seq\":%d, \"cursor\":%lld}",
+             (long long)db_version, seq, (long long)netdata->check_cursor);
 
     NETWORK_RESULT result = network_receive_buffer(netdata, netdata->check_endpoint, netdata->authentication, true, true, json_payload, cloudsync_check_headers, ARRAY_LEN(cloudsync_check_headers));
     int rc = SQLITE_OK;
@@ -1663,20 +1681,36 @@ int cloudsync_network_check_internal(sqlite3_context *context, int *pnrows, sync
             //                              (CHECKPOINT_NONE) so it never lands in
             //                              the middle of a source db_version.
             // The server serves successive chunks of the same stream across /check
-            // calls (the durable cursor stays at "since" until "final"), tracking
-            // chunk-level progress on its side; apply is idempotent, so a stop
-            // before "final" simply re-delivers the stream from "since".
+            // calls (the durable cursor stays at "since" until "final"). The server
+            // is stateless between calls, so it tells us the next spool page via
+            // "cursor" and we echo it back (netdata->check_cursor); apply is
+            // idempotent, so a stop before "final" simply re-delivers from "since".
             int64_t watermark = json_extract_int(result.buffer, result.blen, "watermark", -1);
             int64_t checkpoint_db_version;
             int64_t checkpoint_seq = 0;
+            bool final_chunk = true;
             if (watermark < 0) {
                 checkpoint_db_version = CLOUDSYNC_CHECKPOINT_LAST_APPLIED;
             } else {
-                bool final_chunk = json_extract_bool(result.buffer, result.blen, "final", true);
+                final_chunk = json_extract_bool(result.buffer, result.blen, "final", true);
                 checkpoint_db_version = final_chunk ? watermark : CLOUDSYNC_CHECKPOINT_NONE;
             }
+            // Optional next-page cursor. Absent on legacy/old servers (-1 sentinel)
+            // -> the client does not self-page.
+            int64_t next_cursor = json_extract_int(result.buffer, result.blen, "cursor", -1);
             rc = network_download_changes(context, download_url, pnrows, err_out, checkpoint_db_version, checkpoint_seq);
             cloudsync_memory_free(download_url);
+            if (rc == SQLITE_OK) {
+                // Advance only after the page is applied/staged, mirroring the durable
+                // cursor: a non-final chunk that carried a cursor pages forward;
+                // final or no-cursor ends the drain and resets paging to 0. On a
+                // download/apply failure we leave check_cursor unchanged so a retry
+                // re-requests the same page.
+                netdata->check_cursor = (next_cursor >= 0 && !final_chunk) ? next_cursor : 0;
+            }
+        } else {
+            // 202 / "up to date": no artifact in flight -> reset paging.
+            netdata->check_cursor = 0;
         }
         // failures.check may appear in either shape; extract opportunistically.
         if (out) {

src/postgresql/cloudsync.sql.in

@@ -180,6 +180,97 @@ 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;
+
 -- Payload decoding and application
 CREATE OR REPLACE FUNCTION cloudsync_payload_decode(payload bytea)
 RETURNS integer

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

@@ -34,3 +34,74 @@ CREATE OR REPLACE FUNCTION cloudsync_uuid_blob(uuid text)
 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;

src/sql.h

@@ -74,6 +74,13 @@ 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_CLEANUP_STALE;
+
 // BLOCKS (block-level LWW)
 extern const char * const SQL_BLOCKS_CREATE_TABLE;
 extern const char * const SQL_BLOCKS_UPSERT;