fix(payload): actionable error from cloudsync_payload_chunks when uninitialized (#54)

Querying the cloudsync_payload_chunks vtab on a database where cloudsync
was never initialized surfaced a bare "SQL logic error": the internal
statements over cloudsync_changes failed at step time and the error code
propagated out of xFilter without ever setting the outer vtab's zErrMsg.
The /check endpoint hit this on nodes not configured for cloudsync,
while cloudsync_payload_apply already reported an actionable message.

- guard xFilter with the same is-initialized check (and message) used by
  cloudsync_payload_apply
- propagate inner-statement error messages onto the vtab's zErrMsg
- stop ignoring step failures of the watermark MAX(db_version) query
- unit test: Payload Chunks Uninitialized
- e2e: Failure Path Test now asserts the forwarded "cloudsync is not
  initialized" message on send, receive (non-retryable SQL error), and
  sync; red against nodes still running 1.1.0, green once they run 1.1.1
- bump CLOUDSYNC_VERSION to 1.1.1 and add the CHANGELOG entry (no PG
  changes, extension version stays 1.1 — no migration needed)

Author: andinux

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## [1.1.1] - 2026-07-10
+
+### Fixed
+
+- Querying the SQLite `cloudsync_payload_chunks()` virtual table on a database where cloudsync is not initialized now raises the same actionable error as `cloudsync_payload_apply()` ("cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') ...") instead of a bare "SQL logic error". Errors raised by the internal `cloudsync_changes` scan (including the watermark computation, which previously ignored them) are now propagated with their message instead of being masked.
+
 ## [1.1.0] - 2026-07-09
 
 ### Added

src/cloudsync.h

@@ -18,7 +18,7 @@
 extern "C" {
 #endif
 
-#define CLOUDSYNC_VERSION                       "1.1.0"
+#define CLOUDSYNC_VERSION                       "1.1.1"
 #define CLOUDSYNC_MAX_TABLENAME_LEN             512
 
 #define CLOUDSYNC_VALUE_NOTSET                  -1

src/sqlite/cloudsync_sqlite.c

@@ -1229,7 +1229,12 @@ static int payload_chunks_step_source(cloudsync_payload_chunks_cursor *c) {
     int rc = sqlite3_step(c->src);
     if (rc == SQLITE_ROW) { c->has_row = true; return SQLITE_OK; }
     c->has_row = false;
-    return rc == SQLITE_DONE ? SQLITE_OK : rc;
+    if (rc == SQLITE_DONE) return SQLITE_OK;
+    // copy the inner statement's message onto this vtab or SQLite surfaces the
+    // error as a bare "SQL logic error"
+    if (c->vtab->base.zErrMsg) sqlite3_free(c->vtab->base.zErrMsg);
+    c->vtab->base.zErrMsg = sqlite3_mprintf("%s", sqlite3_errmsg(c->vtab->db));
+    return rc;
 }
 
 static int payload_chunks_plan_fragment(cloudsync_payload_chunks_cursor *c) {
@@ -1427,6 +1432,12 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const
     UNUSED_PARAMETER(idxstr); UNUSED_PARAMETER(argc);
     cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
     cloudsync_context *data = c->vtab->data;
+    if (!cloudsync_context_is_initialized(data)) {
+        c->vtab->base.zErrMsg = sqlite3_mprintf(
+            "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
+            "to enable sync on a table before querying cloudsync_payload_chunks.");
+        return SQLITE_ERROR;
+    }
     if (c->src) { sqlite3_finalize(c->src); c->src = NULL; }
     if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
     // Contract: all per-scan state that can be bulk-reset here must live at or
@@ -1489,8 +1500,16 @@ static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const
         sqlite3_free(mxsql);
         if (rc != SQLITE_OK) return rc;
         sqlite3_bind_blob(mx, 1, site_id, site_id_len, SQLITE_TRANSIENT);
-        if (sqlite3_step(mx) == SQLITE_ROW) until = sqlite3_column_int64(mx, 0);
+        // MAX() yields exactly one row, so anything else is an error: propagate
+        // it instead of silently scanning an empty window with until=0
+        rc = sqlite3_step(mx);
+        if (rc == SQLITE_ROW) until = sqlite3_column_int64(mx, 0);
         sqlite3_finalize(mx);
+        if (rc != SQLITE_ROW) {
+            if (c->vtab->base.zErrMsg) sqlite3_free(c->vtab->base.zErrMsg);
+            c->vtab->base.zErrMsg = sqlite3_mprintf("%s", sqlite3_errmsg(c->vtab->db));
+            return rc == SQLITE_DONE ? SQLITE_ERROR : rc;
+        }
     }
     c->watermark = until;
 

test/integration.c

@@ -1521,16 +1521,33 @@ int test_failure_path (const char *db_path) {
     // Give the server time to process and fail the queued apply/check jobs.
     sqlite3_sleep(5000);
 
-    // Second invocation — failures must surface now.
-    // jobId is always > 0 when failure object is present, so ->> + GT0 doubles as
-    // an existence check (NULL → atoi returns 0 → fails GT0).
+    // Second invocation — failures must surface now. The tenant database is not
+    // configured for cloudsync on the node, so every surfaced failure must carry
+    // the actionable "cloudsync is not initialized" message (instr on the message
+    // doubles as the lastFailure existence check: NULL fails GT0).
     rc = db_expect_gt0(db,
-        "SELECT cloudsync_network_send_changes() ->> '$.send.lastFailure.jobId';"); RCHECK
-    rc = db_expect_gt0(db,
-        "SELECT cloudsync_network_receive_changes() ->> '$.receive.lastFailure.jobId';"); RCHECK
-    // sync must surface at least one of the two; instr() catches either path.
+        "SELECT instr(cloudsync_network_send_changes() ->> '$.send.lastFailure.message', 'cloudsync is not initialized');"); RCHECK
+
+    // The server reports the failed check job as non-retryable, so
+    // receive_changes raises a SQL error instead of returning JSON.
+    char *errmsg = NULL;
+    if (sqlite3_exec(db, "SELECT cloudsync_network_receive_changes();", NULL, NULL, &errmsg) == SQLITE_OK) {
+        printf("Error: expected cloudsync_network_receive_changes to fail, but it succeeded\n");
+        rc = SQLITE_ERROR;
+        goto abort_test;
+    }
+    if (!errmsg || !strstr(errmsg, "cloudsync is not initialized")) {
+        printf("Error: unexpected receive_changes error: %s\n", errmsg ? errmsg : "(null)");
+        sqlite3_free(errmsg);
+        rc = SQLITE_ERROR;
+        goto abort_test;
+    }
+    sqlite3_free(errmsg);
+
+    // sync keeps emitting structured JSON so its send block survives; the
+    // forwarded message must appear in at least one lastFailure object.
     rc = db_expect_gt0(db,
-        "SELECT instr(cloudsync_network_sync(250, 1), '\"lastFailure\":');"); RCHECK
+        "SELECT instr(cloudsync_network_sync(250, 1), 'cloudsync is not initialized');"); RCHECK
 
     rc = db_exec(db, "SELECT cloudsync_terminate();");
 

test/unit.c

@@ -12934,6 +12934,48 @@ bool do_test_payload_chunks_positional_resume (bool print_result, bool cleanup_d
     return result;
 }
 
+// The /check endpoint queries cloudsync_payload_chunks on databases that may not
+// be configured for cloudsync: the vtab must raise the same actionable message
+// cloudsync_payload_apply gives, not a bare "SQL logic error".
+bool do_test_payload_chunks_uninitialized (bool print_result, bool cleanup_databases) {
+    sqlite3 *db = NULL;
+    sqlite3_stmt *stmt = NULL;
+    bool result = false;
+
+    time_t timestamp = time(NULL);
+    int saved_counter = test_counter++;
+
+    db = do_create_database_file(0, timestamp, saved_counter);
+    if (!db) goto finalize;
+
+    // no cloudsync_init on purpose
+    if (sqlite3_prepare_v2(db,
+        "SELECT payload FROM cloudsync_payload_chunks(0, cloudsync_uuid_blob('0190a1b2-c3d4-7e5f-8a9b-001122334455'), NULL, 1) LIMIT 1;",
+        -1, &stmt, NULL) != SQLITE_OK) goto finalize;
+    if (sqlite3_step(stmt) != SQLITE_ERROR) goto finalize;
+    if (!strstr(sqlite3_errmsg(db), "cloudsync is not initialized")) goto finalize;
+    sqlite3_finalize(stmt); stmt = NULL;
+
+    result = true;
+
+finalize:
+    if (!result && print_result) {
+        printf("do_test_payload_chunks_uninitialized error: %s\n", db ? sqlite3_errmsg(db) : "no db");
+    }
+    if (stmt) sqlite3_finalize(stmt);
+    if (db) close_db(db);
+    if (cleanup_databases) {
+        char path[256], walpath[300], shmpath[300];
+        do_build_database_path(path, 0, timestamp, saved_counter);
+        snprintf(walpath, sizeof(walpath), "%s-wal", path);
+        snprintf(shmpath, sizeof(shmpath), "%s-shm", path);
+        file_delete_internal(path);
+        file_delete_internal(walpath);
+        file_delete_internal(shmpath);
+    }
+    return result;
+}
+
 bool do_test_payload_idempotency (int nclients, bool print_result, bool cleanup_databases) {
     sqlite3 *db[2] = {NULL, NULL};
     bool result = false;
@@ -13386,6 +13428,7 @@ int main (int argc, const char * argv[]) {
     result += test_report("Payload Chunks Site Exclusion:", do_test_payload_chunks_site_exclusion(print_result, cleanup_databases));
     result += test_report("Payload Chunks Split db_version:", do_test_payload_chunks_split_dbversion(print_result, cleanup_databases));
     result += test_report("Payload Chunks Positional Resume:", do_test_payload_chunks_positional_resume(print_result, cleanup_databases));
+    result += test_report("Payload Chunks Uninitialized:", do_test_payload_chunks_uninitialized(print_result, cleanup_databases));
 
     // close local database
     close_db(db);