fix: harden checked payload blob cleanup

Document that cloudsync_payload_blob_checked scans successful windows twice so callers understand the I/O tradeoff of guarding monolithic payload allocation.

Add cloudsync_payload_context_free() and use it from SQLite and PostgreSQL checked-blob paths so malloc-backed payload buffers are released on errors and NULL results.

Refactor the PostgreSQL encode pass to reuse the existing PayloadChunksState row extraction helpers, avoiding a duplicated 9-column SPI mapping while preserving the checked window semantics.

Author: andinux

API.md

@@ -528,6 +528,8 @@ This helper is intended for servers that still need to support old clients on a
 
 Internally, CloudSync first estimates the uncompressed payload body plus header. If that estimate exceeds `max_estimated_payload_size`, the function raises a limit-exceeded error and does not materialize the payload. Empty windows return `NULL`.
 
+When the check passes, this function scans the selected change window twice: once to estimate and once to encode. This avoids unsafe monolithic allocation, but successful calls are more I/O-expensive than a direct single-pass `cloudsync_payload_encode()` over the same rows.
+
 **Parameters:**
 
 - `since_db_version` (INTEGER/BIGINT): Start after this source database version, except rows at the same version with `seq > since_seq` are still included.

PERFORMANCE.md

@@ -43,6 +43,8 @@ Each metadata table has an **index on `db_version`**, so payload generation scal
 
 The legacy `cloudsync_payload_encode()` API builds one monolithic LZ4-compressed payload before transmission. For large deltas, `cloudsync_payload_chunks()` can be used instead: it streams a sequence of payload chunks bounded by the `payload_max_chunk_size` setting (default 5 MB, minimum 256 KB). If a single encoded BLOB/TEXT value is larger than the chunk budget, the value is split into transparent v3 fragments and reassembled by `cloudsync_payload_apply()` on the receiver.
 
+For legacy `/check` callers that still need one monolithic payload, `cloudsync_payload_blob_checked()` performs an internal size estimate before encoding. Successful calls scan the selected change window twice (estimate, then encode), so they trade extra I/O for avoiding unsafe monolithic payload allocation when the estimate exceeds the configured limit.
+
 #### Pull: Payload Application
 
 ```

src/cloudsync.c

@@ -3045,6 +3045,12 @@ size_t cloudsync_payload_context_size (size_t *header_size) {
     return sizeof(cloudsync_payload_context);
 }
 
+void cloudsync_payload_context_free (cloudsync_payload_context *payload) {
+    if (!payload) return;
+    if (payload->buffer) cloudsync_memory_free(payload->buffer);
+    cloudsync_memory_free(payload);
+}
+
 uint64_t cloudsync_payload_context_nrows (cloudsync_payload_context *payload) {
     return payload ? payload->nrows : 0;
 }

src/cloudsync.h

@@ -115,6 +115,7 @@ int    cloudsync_payload_encode_step (cloudsync_payload_context *payload, clouds
 int    cloudsync_payload_encode_final (cloudsync_payload_context *payload, cloudsync_context *data);
 char  *cloudsync_payload_blob (cloudsync_payload_context *payload, int64_t *blob_size, int64_t *nrows);
 size_t cloudsync_payload_context_size (size_t *header_size);
+void   cloudsync_payload_context_free (cloudsync_payload_context *payload);
 uint64_t cloudsync_payload_context_nrows (cloudsync_payload_context *payload);
 size_t cloudsync_payload_context_bused (cloudsync_payload_context *payload);
 int    cloudsync_payload_get (cloudsync_context *data, char **blob, int *blob_size, int *db_version, int64_t *new_db_version);

src/postgresql/cloudsync_postgresql.c

@@ -1480,6 +1480,9 @@ PG_FUNCTION_INFO_V1(cloudsync_payload_blob_checked);
 Datum cloudsync_payload_blob_checked(PG_FUNCTION_ARGS) {
     cloudsync_context *data = get_cloudsync_context();
     bool spi_connected = false;
+    Portal portal = NULL;
+    PayloadChunksState encode_st = {0};
+    cloudsync_payload_context *payload = NULL;
     bytea *site_id = PG_ARGISNULL(2) ? NULL : PG_GETARG_BYTEA_PP(2);
     bool exclude = PG_ARGISNULL(3) ? false : PG_GETARG_BOOL(3);
 
@@ -1520,100 +1523,79 @@ Datum cloudsync_payload_blob_checked(PG_FUNCTION_ARGS) {
                    (long long)estimated, (long long)max_estimated_size)));
     }
 
-    StringInfoData q;
-    initStringInfo(&q);
-    if (exclude) {
-        appendStringInfoString(&q,
-            "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
-            "FROM cloudsync_changes_select(0,NULL) "
-            "WHERE site_id <> $2 AND db_version <= $3 AND (db_version > $1 OR (db_version = $1 AND seq > $4)) "
-            "ORDER BY db_version, seq ASC");
-    } else {
-        appendStringInfoString(&q,
-            "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
-            "FROM cloudsync_changes_select(0,$2) "
-            "WHERE db_version <= $3 AND (db_version > $1 OR (db_version = $1 AND seq > $4)) "
-            "ORDER BY db_version, seq ASC");
-    }
-    Oid argtypes[4] = {INT8OID, BYTEAOID, INT8OID, INT8OID};
-    Datum values[4] = {Int64GetDatum(since), PointerGetDatum(site_id), Int64GetDatum(watermark), Int64GetDatum(since_seq)};
-    char nulls[4] = {' ', ' ', ' ', ' '};
-    Portal portal = SPI_cursor_open_with_args(NULL, q.data, 4, argtypes, values, nulls, true, 0);
-    pfree(q.data);
-    if (!portal) ereport(ERROR, (errmsg("SPI_cursor_open failed")));
+    PG_TRY();
+    {
+        StringInfoData q;
+        initStringInfo(&q);
+        if (exclude) {
+            appendStringInfoString(&q,
+                "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
+                "FROM cloudsync_changes_select(0,NULL) "
+                "WHERE site_id <> $2 AND db_version <= $3 AND (db_version > $1 OR (db_version = $1 AND seq > $4)) "
+                "ORDER BY db_version, seq ASC");
+        } else {
+            appendStringInfoString(&q,
+                "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
+                "FROM cloudsync_changes_select(0,$2) "
+                "WHERE db_version <= $3 AND (db_version > $1 OR (db_version = $1 AND seq > $4)) "
+                "ORDER BY db_version, seq ASC");
+        }
+        Oid argtypes[4] = {INT8OID, BYTEAOID, INT8OID, INT8OID};
+        Datum values[4] = {Int64GetDatum(since), PointerGetDatum(site_id), Int64GetDatum(watermark), Int64GetDatum(since_seq)};
+        char nulls[4] = {' ', ' ', ' ', ' '};
+        portal = SPI_cursor_open_with_args(NULL, q.data, 4, argtypes, values, nulls, true, 0);
+        pfree(q.data);
+        if (!portal) ereport(ERROR, (errmsg("SPI_cursor_open failed")));
+        encode_st.portal = portal;
 
-    cloudsync_payload_context *payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
-    if (!payload) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory")));
+        payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
+        if (!payload) ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory")));
 
-    for (;;) {
-        SPI_cursor_fetch(portal, true, 1);
-        if (SPI_processed == 0) {
-            if (SPI_tuptable) { SPI_freetuptable(SPI_tuptable); SPI_tuptable = NULL; }
-            break;
+        while (payload_chunks_fetch_current(&encode_st)) {
+            pgvalue_t *vals[9] = {0};
+            text *owned_texts[2] = {0};
+            payload_chunks_make_pgvalues(&encode_st, vals, owned_texts);
+            int rc = cloudsync_payload_encode_step(payload, data, 9, (dbvalue_t **)vals);
+            payload_chunks_free_pgvalues(vals, owned_texts);
+            if (rc != DBRES_OK) {
+                ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", cloudsync_errmsg(data))));
+            }
+            payload_chunks_free_current(&encode_st);
         }
-        SPITupleTable *tuptable = SPI_tuptable;
-        HeapTuple tup = tuptable->vals[0];
-        TupleDesc td = tuptable->tupdesc;
-        bool isnull = false;
-        Datum d;
-
-        char *tbl = NULL;
-        bytea *pk = NULL;
-        char *col_name = NULL;
-        bytea *col_value = NULL;
-        bytea *row_site_id = NULL;
-        int64 col_version = 0, db_version = 0, cl = 0, seq = 0;
-
-        d = SPI_getbinval(tup, td, 1, &isnull); tbl = isnull ? pstrdup("") : text_to_cstring(DatumGetTextPP(d));
-        d = SPI_getbinval(tup, td, 2, &isnull); if (!isnull) pk = DatumGetByteaPP(d);
-        d = SPI_getbinval(tup, td, 3, &isnull); col_name = isnull ? pstrdup("") : text_to_cstring(DatumGetTextPP(d));
-        d = SPI_getbinval(tup, td, 4, &isnull); if (!isnull) col_value = DatumGetByteaPP(d);
-        d = SPI_getbinval(tup, td, 5, &isnull); col_version = isnull ? 0 : DatumGetInt64(d);
-        d = SPI_getbinval(tup, td, 6, &isnull); db_version = isnull ? 0 : DatumGetInt64(d);
-        d = SPI_getbinval(tup, td, 7, &isnull); if (!isnull) row_site_id = DatumGetByteaPP(d);
-        d = SPI_getbinval(tup, td, 8, &isnull); cl = isnull ? 0 : DatumGetInt64(d);
-        d = SPI_getbinval(tup, td, 9, &isnull); seq = isnull ? 0 : DatumGetInt64(d);
+        SPI_cursor_close(portal);
+        portal = NULL;
+        encode_st.portal = NULL;
 
-        pgvalue_t *vals[9] = {0};
-        text *owned_texts[2] = {0};
-        owned_texts[0] = cstring_to_text(tbl);
-        owned_texts[1] = cstring_to_text(col_name);
-        vals[0] = pgvalue_create(PointerGetDatum(owned_texts[0]), TEXTOID, -1, InvalidOid, false);
-        vals[1] = pgvalue_create(PointerGetDatum(pk), BYTEAOID, -1, InvalidOid, false);
-        vals[2] = pgvalue_create(PointerGetDatum(owned_texts[1]), TEXTOID, -1, InvalidOid, false);
-        vals[3] = pgvalue_create(PointerGetDatum(col_value), BYTEAOID, -1, InvalidOid, false);
-        vals[4] = pgvalue_create(Int64GetDatum(col_version), INT8OID, -1, InvalidOid, false);
-        vals[5] = pgvalue_create(Int64GetDatum(db_version), INT8OID, -1, InvalidOid, false);
-        vals[6] = pgvalue_create(PointerGetDatum(row_site_id), BYTEAOID, -1, InvalidOid, false);
-        vals[7] = pgvalue_create(Int64GetDatum(cl), INT8OID, -1, InvalidOid, false);
-        vals[8] = pgvalue_create(Int64GetDatum(seq), INT8OID, -1, InvalidOid, false);
-        int rc = cloudsync_payload_encode_step(payload, data, 9, (dbvalue_t **)vals);
-        payload_chunks_free_pgvalues(vals, owned_texts);
-        pfree(tbl);
-        pfree(col_name);
+        int rc = cloudsync_payload_encode_final(payload, data);
         if (rc != DBRES_OK) ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", cloudsync_errmsg(data))));
+        int64 blob_size = 0;
+        char *blob = cloudsync_payload_blob(payload, &blob_size, NULL);
+        if (spi_connected) {
+            SPI_finish();
+            spi_connected = false;
+        }
+        if (!blob) {
+            cloudsync_payload_context_free(payload);
+            payload = NULL;
+            PG_RETURN_NULL();
+        }
 
-        if (tuptable) SPI_freetuptable(tuptable);
-        SPI_tuptable = NULL;
+        bytea *result = (bytea *)palloc(VARHDRSZ + blob_size);
+        SET_VARSIZE(result, VARHDRSZ + blob_size);
+        memcpy(VARDATA(result), blob, blob_size);
+        cloudsync_payload_context_free(payload);
+        payload = NULL;
+        PG_RETURN_BYTEA_P(result);
     }
-    SPI_cursor_close(portal);
-
-    int rc = cloudsync_payload_encode_final(payload, data);
-    if (rc != DBRES_OK) ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", cloudsync_errmsg(data))));
-    int64 blob_size = 0;
-    char *blob = cloudsync_payload_blob(payload, &blob_size, NULL);
-    if (spi_connected) SPI_finish();
-    if (!blob) {
-        cloudsync_memory_free(payload);
-        PG_RETURN_NULL();
+    PG_CATCH();
+    {
+        if (portal) SPI_cursor_close(portal);
+        payload_chunks_free_current(&encode_st);
+        if (payload) cloudsync_payload_context_free(payload);
+        if (spi_connected) SPI_finish();
+        PG_RE_THROW();
     }
-
-    bytea *result = (bytea *)palloc(VARHDRSZ + blob_size);
-    SET_VARSIZE(result, VARHDRSZ + blob_size);
-    memcpy(VARDATA(result), blob, blob_size);
-    cloudsync_memory_free(blob);
-    cloudsync_memory_free(payload);
-    PG_RETURN_BYTEA_P(result);
+    PG_END_TRY();
 }
 
 // Payload decode - Apply changes from payload

src/sqlite/cloudsync_sqlite.c

@@ -1508,6 +1508,7 @@ static int payload_blob_checked_estimate(sqlite3 *db, const void *site_id, int s
     int rc = SQLITE_OK;
     sqlite3_int64 until = 0;
     size_t header_size = 0;
+    bool has_rows = false;
     const char *site_op = exclude ? "<>" : "=";
     *estimated = 0;
 
@@ -1551,13 +1552,14 @@ static int payload_blob_checked_estimate(sqlite3 *db, const void *site_id, int s
             rc = SQLITE_TOOBIG;
             goto error;
         }
-        if (*estimated == 0) {
+        if (!has_rows) {
             if (header_size > (size_t)INT64_MAX) {
                 rc = SQLITE_TOOBIG;
                 goto error;
             }
             rc = payload_estimated_size_add(estimated, (sqlite3_int64)header_size);
             if (rc != SQLITE_OK) goto error;
+            has_rows = true;
         }
         rc = payload_estimated_size_add(estimated, (sqlite3_int64)row_size);
         if (rc != SQLITE_OK) goto error;
@@ -1661,15 +1663,16 @@ void dbsync_payload_blob_checked(sqlite3_context *context, int argc, sqlite3_val
     char *blob = cloudsync_payload_blob(payload, &blob_size, NULL);
     if (!blob) {
         sqlite3_result_null(context);
+        cloudsync_payload_context_free(payload);
     } else {
         sqlite3_result_blob64(context, blob, (sqlite3_uint64)blob_size, cloudsync_memory_free);
+        cloudsync_memory_free(payload);
     }
-    cloudsync_memory_free(payload);
     return;
 
 error:
     if (stmt) sqlite3_finalize(stmt);
-    if (payload) cloudsync_memory_free(payload);
+    if (payload) cloudsync_payload_context_free(payload);
     if (rc == SQLITE_NOMEM) sqlite3_result_error_nomem(context);
     else if (rc == SQLITE_TOOBIG) sqlite3_result_error(context, "cloudsync_payload_blob_checked: payload estimate is too large", -1);
     else sqlite3_result_error(context, sqlite3_errmsg(db), -1);