Line data Source code
1 : //
2 : // cloudsync.c
3 : // cloudsync
4 : //
5 : // Created by Marco Bambini on 16/05/24.
6 : //
7 :
8 : #include <inttypes.h>
9 : #include <stdbool.h>
10 : #include <limits.h>
11 : #include <stdint.h>
12 : #include <stdlib.h>
13 : #include <string.h>
14 : #include <assert.h>
15 : #include <stdio.h>
16 : #include <errno.h>
17 : #include <math.h>
18 : #include <time.h>
19 :
20 : #include "cloudsync.h"
21 : #include "lz4.h"
22 : #include "pk.h"
23 : #include "sql.h"
24 : #include "utils.h"
25 : #include "dbutils.h"
26 : #include "block.h"
27 :
28 : #ifdef _WIN32
29 : #include <winsock2.h>
30 : #include <ws2tcpip.h>
31 : #else
32 : #include <arpa/inet.h> // for htonl, htons, ntohl, ntohs
33 : #include <netinet/in.h> // for struct sockaddr_in, INADDR_ANY, etc. (if needed)
34 : #endif
35 :
36 : #ifndef htonll
37 : #if __BIG_ENDIAN__
38 : #define htonll(x) (x)
39 : #define ntohll(x) (x)
40 : #else
41 : #ifndef htobe64
42 : #define htonll(x) ((uint64_t)htonl((x) & 0xFFFFFFFF) << 32 | (uint64_t)htonl((x) >> 32))
43 : #define ntohll(x) ((uint64_t)ntohl((x) & 0xFFFFFFFF) << 32 | (uint64_t)ntohl((x) >> 32))
44 : #else
45 : #define htonll(x) htobe64(x)
46 : #define ntohll(x) be64toh(x)
47 : #endif
48 : #endif
49 : #endif
50 :
51 : #define CLOUDSYNC_INIT_NTABLES 64
52 : #define CLOUDSYNC_MIN_DB_VERSION 0
53 :
54 : #define CLOUDSYNC_PAYLOAD_MINBUF_SIZE (512*1024)
55 : #define CLOUDSYNC_PAYLOAD_SIGNATURE 0x434C5359 /* 'C','L','S','Y' */
56 : #define CLOUDSYNC_PAYLOAD_VERSION_ORIGNAL 1
57 : #define CLOUDSYNC_PAYLOAD_VERSION_1 CLOUDSYNC_PAYLOAD_VERSION_ORIGNAL
58 : #define CLOUDSYNC_PAYLOAD_VERSION_2 2
59 : #define CLOUDSYNC_PAYLOAD_VERSION_3 3
60 : #define CLOUDSYNC_PAYLOAD_VERSION_LATEST CLOUDSYNC_PAYLOAD_VERSION_2
61 : #define CLOUDSYNC_PAYLOAD_MIN_VERSION_WITH_CHECKSUM CLOUDSYNC_PAYLOAD_VERSION_2
62 : #define CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX "__cloudsync_frag_v1__:"
63 : #define CLOUDSYNC_PAYLOAD_FRAGMENT_STALE_SECONDS (24*60*60)
64 : #define CLOUDSYNC_PAYLOAD_FRAGMENT_CLEANUP_MIN_INTERVAL (60)
65 :
66 : #ifndef MAX
67 : #define MAX(a, b) (((a)>(b))?(a):(b))
68 : #endif
69 :
70 : #define DEBUG_DBERROR(_rc, _fn, _data) do {if (_rc != DBRES_OK) printf("Error in %s: %s\n", _fn, database_errmsg(_data));} while (0)
71 :
72 : typedef enum {
73 : CLOUDSYNC_PK_INDEX_TBL = 0,
74 : CLOUDSYNC_PK_INDEX_PK = 1,
75 : CLOUDSYNC_PK_INDEX_COLNAME = 2,
76 : CLOUDSYNC_PK_INDEX_COLVALUE = 3,
77 : CLOUDSYNC_PK_INDEX_COLVERSION = 4,
78 : CLOUDSYNC_PK_INDEX_DBVERSION = 5,
79 : CLOUDSYNC_PK_INDEX_SITEID = 6,
80 : CLOUDSYNC_PK_INDEX_CL = 7,
81 : CLOUDSYNC_PK_INDEX_SEQ = 8
82 : } CLOUDSYNC_PK_INDEX;
83 :
84 : typedef enum {
85 : DBVM_VALUE_ERROR = -1,
86 : DBVM_VALUE_UNCHANGED = 0,
87 : DBVM_VALUE_CHANGED = 1,
88 : } DBVM_VALUE;
89 :
90 : #define SYNCBIT_SET(_data) _data->insync = 1
91 : #define SYNCBIT_RESET(_data) _data->insync = 0
92 :
93 : // MARK: - Deferred column-batch merge -
94 :
95 : typedef struct {
96 : const char *col_name; // pointer into table_context->col_name[idx] (stable)
97 : dbvalue_t *col_value; // duplicated via database_value_dup (owned)
98 : int64_t col_version;
99 : int64_t db_version;
100 : uint8_t site_id[UUID_LEN];
101 : int site_id_len;
102 : int64_t seq;
103 : } merge_pending_entry;
104 :
105 : typedef struct {
106 : cloudsync_table_context *table;
107 : char *pk; // malloc'd copy, freed on flush
108 : int pk_len;
109 : int64_t cl;
110 : bool sentinel_pending;
111 : bool row_exists; // true when the PK already exists locally
112 : int count;
113 : int capacity;
114 : merge_pending_entry *entries;
115 :
116 : // Statement cache — reuse the prepared statement when the column
117 : // combination and row_exists flag match between consecutive PK flushes.
118 : dbvm_t *cached_vm;
119 : bool cached_row_exists;
120 : int cached_col_count;
121 : const char **cached_col_names; // array of pointers into table_context (not owned)
122 :
123 : // True while cloudsync_payload_apply holds a savepoint around the whole PK group: the
124 : // flush then writes inside it instead of opening its own, so that a failed flush also
125 : // rolls back what the group's rows already wrote (a resurrected row's sentinel and
126 : // zeroed clocks, block values, winner clocks).
127 : bool group_savepoint;
128 : } merge_pending_batch;
129 :
130 : // MARK: -
131 :
132 : struct cloudsync_pk_decode_bind_context {
133 : dbvm_t *vm;
134 : char *tbl;
135 : int64_t tbl_len;
136 : const void *pk;
137 : int64_t pk_len;
138 : char *col_name;
139 : int64_t col_name_len;
140 : int64_t col_version;
141 : int64_t db_version;
142 : const void *site_id;
143 : int64_t site_id_len;
144 : int64_t cl;
145 : int64_t seq;
146 : };
147 :
148 : typedef struct {
149 : char value_id[33];
150 : bool applied; // applied by this stream: later pieces are redundant
151 : } cloudsync_stream_value;
152 :
153 : struct cloudsync_context {
154 : void *db;
155 : char errmsg[1024];
156 : int errcode;
157 : int sqlstate; // SQLSTATE behind errmsg (PostgreSQL), 0 if none
158 :
159 : char *libversion;
160 : uint8_t site_id[UUID_LEN];
161 : int insync;
162 : int debug;
163 : bool merge_equal_values;
164 : void *aux_data;
165 :
166 : // stmts and context values
167 : dbvm_t *schema_version_stmt;
168 : dbvm_t *data_version_stmt;
169 : dbvm_t *db_version_stmt;
170 : dbvm_t *getset_siteid_stmt;
171 : int data_version;
172 : int schema_version;
173 : uint64_t schema_hash;
174 :
175 : // set at transaction start and reset on commit/rollback
176 : int64_t db_version;
177 : // version the DB would have if the transaction committed now
178 : int64_t pending_db_version;
179 : // set when a payload group ends inside a transaction: the next change takes a new version
180 : bool pending_closed;
181 : // used to set an order inside each transaction
182 : int seq;
183 :
184 : // wall-clock (time()) of the last stale v3-fragment GC; throttles the GC so
185 : // it does not run a full table scan on every applied fragment (0 = never run)
186 : int64_t last_fragment_cleanup;
187 :
188 : // optional schema_name to be set in the cloudsync_table_context
189 : char *current_schema;
190 :
191 : // augmented tables are stored in-memory so we do not need to retrieve information about
192 : // col_names and cid from the disk each time a write statement is performed
193 : // we do also not need to use an hash map here because for few tables the direct
194 : // in-memory comparison with table name is faster
195 : cloudsync_table_context **tables; // dense vector: [0..tables_count-1] are valid
196 : int tables_count; // size
197 : int tables_cap; // capacity
198 :
199 : int skip_decode_idx; // -1 in sqlite, col_value index in postgresql
200 :
201 : // deferred column-batch merge (active during payload_apply)
202 : merge_pending_batch *pending_batch;
203 :
204 : // last (db_version, seq) successfully applied during the current
205 : // cloudsync_payload_apply call; used to resolve the
206 : // CLOUDSYNC_CHECKPOINT_LAST_APPLIED receive-checkpoint mode (-1 = none yet).
207 : int64_t apply_last_db_version;
208 : int64_t apply_last_seq;
209 :
210 : // Entries applied, accumulated across a receive drain so an early chunk is still
211 : // visible when a later chunk reports, including the changes a failing payload
212 : // applied before its error. Reset with cloudsync_apply_stats_reset.
213 : int apply_rows;
214 :
215 : // v3 values the current receive stream delivered: pending ones are checked before
216 : // the stream's final checkpoint. See cloudsync_receive_stream_reset.
217 : cloudsync_stream_value *stream_values;
218 : int stream_values_count;
219 : int stream_values_cap;
220 : };
221 :
222 : struct cloudsync_table_context {
223 : table_algo algo; // CRDT algoritm associated to the table
224 : char *name; // table name
225 : char *schema; // table schema
226 : char *meta_ref; // schema-qualified meta table name (e.g. "schema"."name_cloudsync")
227 : char *base_ref; // schema-qualified base table name (e.g. "schema"."name")
228 : char **col_name; // array of column names
229 : dbvm_t **col_merge_stmt; // array of merge insert stmt (indexed by col_name)
230 : dbvm_t **col_update_stmt; // lazily prepared UPDATE of one column by pk (block materialization)
231 : dbvm_t **col_value_stmt; // array of column value stmt (indexed by col_name)
232 : int *col_id; // array of column id
233 : col_algo_t *col_algo; // per-column algorithm (normal or block)
234 : char **col_delimiter; // per-column delimiter for block splitting (NULL = default "\n")
235 : bool has_block_cols; // quick check: does this table have any block columns?
236 : dbvm_t *block_value_read_stmt; // SELECT col_value FROM blocks table
237 : dbvm_t *block_value_write_stmt; // INSERT OR REPLACE into blocks table
238 : dbvm_t *block_value_delete_stmt; // DELETE from blocks table
239 : dbvm_t *block_list_stmt; // SELECT block entries for materialization
240 : char *blocks_ref; // schema-qualified blocks table name
241 : int ncols; // number of non primary key cols
242 : int npks; // number of primary key cols
243 : bool enabled; // flag to check if a table is enabled or disabled
244 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
245 : bool rowid_only; // a table with no primary keys other than the implicit rowid
246 : #endif
247 :
248 : char **pk_name; // array of primary key names
249 :
250 : // precompiled statements
251 : dbvm_t *meta_pkexists_stmt; // check if a primary key already exist in the augmented table
252 : dbvm_t *meta_sentinel_update_stmt; // update a local sentinel row
253 : dbvm_t *meta_sentinel_insert_stmt; // insert a local sentinel row
254 : dbvm_t *meta_row_insert_update_stmt; // insert/update a local row
255 : dbvm_t *meta_row_drop_stmt; // delete rows from meta
256 : dbvm_t *meta_update_move_stmt; // update rows in meta when pk changes
257 : dbvm_t *meta_local_cl_stmt; // compute local cl value
258 : dbvm_t *meta_winner_clock_stmt; // get the rowid of the last inserted/updated row in the meta table
259 : dbvm_t *meta_merge_delete_drop;
260 : dbvm_t *meta_zero_clock_stmt;
261 : dbvm_t *meta_col_version_stmt;
262 : dbvm_t *meta_site_id_stmt;
263 :
264 : dbvm_t *real_col_values_stmt; // retrieve all column values based on pk
265 : dbvm_t *real_merge_delete_stmt;
266 : dbvm_t *real_merge_sentinel_stmt;
267 :
268 : bool is_altering; // flag to track if a table alteration is in progress
269 :
270 : // context
271 : cloudsync_context *context;
272 : };
273 :
274 : struct cloudsync_payload_context {
275 : char *buffer;
276 : size_t bsize;
277 : size_t balloc;
278 : size_t bused;
279 : uint64_t nrows;
280 : uint16_t ncols;
281 : uint8_t version;
282 : };
283 :
284 : #ifdef _MSC_VER
285 : #pragma pack(push, 1) // For MSVC: pack struct with 1-byte alignment
286 : #define PACKED
287 : #else
288 : #define PACKED __attribute__((__packed__))
289 : #endif
290 :
291 : typedef struct PACKED {
292 : uint32_t signature; // 'CLSY'
293 : uint8_t version; // protocol version
294 : uint8_t libversion[3]; // major.minor.patch
295 : uint32_t expanded_size;
296 : uint16_t ncols;
297 : uint32_t nrows;
298 : uint64_t schema_hash;
299 : uint8_t checksum[6]; // 48 bits checksum (to ensure struct is 32 bytes)
300 : } cloudsync_payload_header;
301 :
302 : #ifdef _MSC_VER
303 : #pragma pack(pop)
304 : #endif
305 :
306 : #if CLOUDSYNC_UNITTEST
307 : bool force_uncompressed_blob = false;
308 : #define CHECK_FORCE_UNCOMPRESSED_BUFFER() if (force_uncompressed_blob) use_uncompressed_buffer = true
309 : #else
310 : #define CHECK_FORCE_UNCOMPRESSED_BUFFER()
311 : #endif
312 :
313 : // Internal prototypes
314 : int local_mark_insert_or_update_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const char *col_name, int64_t db_version, int seq);
315 :
316 : // MARK: - CRDT algos -
317 :
318 1702 : table_algo cloudsync_algo_from_name (const char *algo_name) {
319 1702 : if (algo_name == NULL) return table_algo_none;
320 :
321 1702 : if ((strcasecmp(algo_name, "CausalLengthSet") == 0) || (strcasecmp(algo_name, "cls") == 0)) return table_algo_crdt_cls;
322 800 : if ((strcasecmp(algo_name, "GrowOnlySet") == 0) || (strcasecmp(algo_name, "gos") == 0)) return table_algo_crdt_gos;
323 793 : if ((strcasecmp(algo_name, "DeleteWinsSet") == 0) || (strcasecmp(algo_name, "dws") == 0)) return table_algo_crdt_dws;
324 792 : if ((strcasecmp(algo_name, "AddWinsSet") == 0) || (strcasecmp(algo_name, "aws") == 0)) return table_algo_crdt_aws;
325 :
326 : // if nothing is found
327 791 : return table_algo_none;
328 1702 : }
329 :
330 110 : const char *cloudsync_algo_name (table_algo algo) {
331 110 : switch (algo) {
332 105 : case table_algo_crdt_cls: return "cls";
333 1 : case table_algo_crdt_gos: return "gos";
334 1 : case table_algo_crdt_dws: return "dws";
335 1 : case table_algo_crdt_aws: return "aws";
336 1 : case table_algo_none: return NULL;
337 : }
338 1 : return NULL;
339 110 : }
340 :
341 : // MARK: - DBVM Utils -
342 :
343 46521 : DBVM_VALUE dbvm_execute (dbvm_t *stmt, cloudsync_context *data) {
344 46521 : if (!stmt) return DBVM_VALUE_ERROR;
345 :
346 46519 : int rc = databasevm_step(stmt);
347 46519 : if (rc != DBRES_ROW && rc != DBRES_DONE) {
348 2 : if (data) DEBUG_DBERROR(rc, "stmt_execute", data);
349 2 : databasevm_reset(stmt);
350 2 : return DBVM_VALUE_ERROR;
351 : }
352 :
353 46517 : DBVM_VALUE result = DBVM_VALUE_CHANGED;
354 46517 : if (stmt == data->data_version_stmt) {
355 40241 : int version = (int)database_column_int(stmt, 0);
356 40241 : if (version != data->data_version) {
357 764 : data->data_version = version;
358 764 : } else {
359 39477 : result = DBVM_VALUE_UNCHANGED;
360 : }
361 46517 : } else if (stmt == data->schema_version_stmt) {
362 3138 : int version = (int)database_column_int(stmt, 0);
363 3138 : if (version > data->schema_version) {
364 1631 : data->schema_version = version;
365 1631 : } else {
366 1507 : result = DBVM_VALUE_UNCHANGED;
367 : }
368 :
369 6276 : } else if (stmt == data->db_version_stmt) {
370 3138 : data->db_version = (rc == DBRES_DONE) ? CLOUDSYNC_MIN_DB_VERSION : database_column_int(stmt, 0);
371 3138 : }
372 :
373 46517 : databasevm_reset(stmt);
374 46517 : return result;
375 46521 : }
376 :
377 6013 : int dbvm_count (dbvm_t *stmt, const char *value, size_t len, int type) {
378 6013 : int result = -1;
379 6013 : int rc = DBRES_OK;
380 :
381 6013 : if (value) {
382 6012 : rc = (type == DBTYPE_TEXT) ? databasevm_bind_text(stmt, 1, value, (int)len) : databasevm_bind_blob(stmt, 1, value, len);
383 6012 : if (rc != DBRES_OK) goto cleanup;
384 6012 : }
385 :
386 6013 : rc = databasevm_step(stmt);
387 12026 : if (rc == DBRES_DONE) {
388 1 : result = 0;
389 1 : rc = DBRES_OK;
390 6013 : } else if (rc == DBRES_ROW) {
391 6012 : result = (int)database_column_int(stmt, 0);
392 6012 : rc = DBRES_OK;
393 6012 : }
394 :
395 : cleanup:
396 6013 : databasevm_reset(stmt);
397 6013 : return result;
398 : }
399 :
400 307060 : void dbvm_reset (dbvm_t *stmt) {
401 307060 : if (!stmt) return;
402 281277 : databasevm_clear_bindings(stmt);
403 281277 : databasevm_reset(stmt);
404 307060 : }
405 :
406 : // MARK: - Database Version -
407 :
408 2465 : int cloudsync_dbversion_build_query (cloudsync_context *data, char **sql_out) {
409 : // this function must be manually called each time tables changes
410 : // because the query plan changes too and it must be re-prepared
411 : // unfortunately there is no other way
412 :
413 : // we need to execute a query like:
414 : /*
415 : SELECT max(version) as version FROM (
416 : SELECT max(db_version) as version FROM "table1_cloudsync"
417 : UNION ALL
418 : SELECT max(db_version) as version FROM "table2_cloudsync"
419 : UNION ALL
420 : SELECT max(db_version) as version FROM "table3_cloudsync"
421 : UNION
422 : SELECT value as version FROM cloudsync_settings WHERE key = 'pre_alter_dbversion'
423 : )
424 : */
425 :
426 : // the good news is that the query can be computed in SQLite without the need to do any extra computation from the host language
427 :
428 2465 : *sql_out = NULL;
429 2465 : return database_select_text(data, SQL_DBVERSION_BUILD_QUERY, sql_out);
430 : }
431 :
432 3222 : int cloudsync_dbversion_rebuild (cloudsync_context *data) {
433 3222 : if (data->db_version_stmt) {
434 1697 : databasevm_finalize(data->db_version_stmt);
435 1697 : data->db_version_stmt = NULL;
436 1697 : }
437 :
438 3222 : int64_t count = dbutils_table_settings_count_tables(data);
439 3222 : if (count == 0) return DBRES_OK;
440 2465 : else if (count == -1) return cloudsync_set_dberror(data);
441 :
442 2465 : char *sql = NULL;
443 2465 : int rc = cloudsync_dbversion_build_query(data, &sql);
444 2465 : if (rc != DBRES_OK) return cloudsync_set_dberror(data);
445 :
446 : // A NULL SQL with rc == OK means the generator produced a NULL row:
447 : // sqlite_master has no *_cloudsync meta-tables (for example, the user
448 : // dropped the base table and its meta-table without calling
449 : // cloudsync_cleanup, leaving stale cloudsync_table_settings rows).
450 : // Treat this the same as count == 0: no prepared statement, db_version
451 : // stays at the minimum and will be rebuilt on the next cloudsync_init.
452 : // Genuine errors from database_select_text are handled above.
453 2464 : if (!sql) return DBRES_OK;
454 : DEBUG_SQL("db_version_stmt: %s", sql);
455 :
456 2463 : rc = databasevm_prepare(data, sql, (void **)&data->db_version_stmt, DBFLAG_PERSISTENT);
457 : DEBUG_STMT("db_version_stmt %p", data->db_version_stmt);
458 2463 : cloudsync_memory_free(sql);
459 2463 : return rc;
460 3222 : }
461 :
462 3138 : int cloudsync_dbversion_rerun (cloudsync_context *data) {
463 3138 : DBVM_VALUE schema_changed = dbvm_execute(data->schema_version_stmt, data);
464 3138 : if (schema_changed == DBVM_VALUE_ERROR) return -1;
465 :
466 3138 : if (schema_changed == DBVM_VALUE_CHANGED) {
467 1631 : int rc = cloudsync_dbversion_rebuild(data);
468 1631 : if (rc != DBRES_OK) return -1;
469 1631 : }
470 :
471 3138 : if (!data->db_version_stmt) {
472 0 : data->db_version = CLOUDSYNC_MIN_DB_VERSION;
473 0 : return 0;
474 : }
475 :
476 3138 : DBVM_VALUE rc = dbvm_execute(data->db_version_stmt, data);
477 3138 : if (rc == DBVM_VALUE_ERROR) return -1;
478 3138 : return 0;
479 3138 : }
480 :
481 40243 : int cloudsync_dbversion_check_uptodate (cloudsync_context *data) {
482 : // perform a PRAGMA data_version to check if some other process write any data
483 40243 : DBVM_VALUE rc = dbvm_execute(data->data_version_stmt, data);
484 40243 : if (rc == DBVM_VALUE_ERROR) return -1;
485 :
486 : // db_version is already set and there is no need to update it
487 40241 : if (data->db_version != CLOUDSYNC_VALUE_NOTSET && rc == DBVM_VALUE_UNCHANGED) return 0;
488 :
489 3138 : return cloudsync_dbversion_rerun(data);
490 40243 : }
491 :
492 40214 : int64_t cloudsync_dbversion_next (cloudsync_context *data, int64_t merging_version) {
493 40214 : int rc = cloudsync_dbversion_check_uptodate(data);
494 40214 : if (rc != DBRES_OK) return -1;
495 :
496 40213 : int64_t result = data->db_version + 1;
497 40213 : if (result < data->pending_db_version) result = data->pending_db_version;
498 40213 : if (data->pending_closed && result <= data->pending_db_version) result = data->pending_db_version + 1;
499 40213 : data->pending_closed = false;
500 40213 : if (merging_version != CLOUDSYNC_VALUE_NOTSET && result < merging_version) result = merging_version;
501 40213 : data->pending_db_version = result;
502 :
503 40213 : return result;
504 40214 : }
505 :
506 : // MARK: - PK Context -
507 :
508 0 : char *cloudsync_pk_context_tbl (cloudsync_pk_decode_bind_context *ctx, int64_t *tbl_len) {
509 0 : *tbl_len = ctx->tbl_len;
510 0 : return ctx->tbl;
511 : }
512 :
513 0 : void *cloudsync_pk_context_pk (cloudsync_pk_decode_bind_context *ctx, int64_t *pk_len) {
514 0 : *pk_len = ctx->pk_len;
515 0 : return (void *)ctx->pk;
516 : }
517 :
518 0 : char *cloudsync_pk_context_colname (cloudsync_pk_decode_bind_context *ctx, int64_t *colname_len) {
519 0 : *colname_len = ctx->col_name_len;
520 0 : return ctx->col_name;
521 : }
522 :
523 0 : int64_t cloudsync_pk_context_cl (cloudsync_pk_decode_bind_context *ctx) {
524 0 : return ctx->cl;
525 : }
526 :
527 0 : int64_t cloudsync_pk_context_dbversion (cloudsync_pk_decode_bind_context *ctx) {
528 0 : return ctx->db_version;
529 : }
530 :
531 : // MARK: - CloudSync Context -
532 :
533 21167 : int cloudsync_insync (cloudsync_context *data) {
534 21167 : return data->insync;
535 : }
536 :
537 1531 : void *cloudsync_siteid (cloudsync_context *data) {
538 1531 : return (void *)data->site_id;
539 : }
540 :
541 2 : void cloudsync_reset_siteid (cloudsync_context *data) {
542 2 : memset(data->site_id, 0, sizeof(uint8_t) * UUID_LEN);
543 2 : }
544 :
545 769 : int cloudsync_load_siteid (cloudsync_context *data) {
546 : // check if site_id was already loaded
547 769 : if (data->site_id[0] != 0) return DBRES_OK;
548 :
549 : // load site_id
550 767 : char *buffer = NULL;
551 767 : int64_t size = 0;
552 767 : int rc = database_select_blob(data, SQL_SITEID_SELECT_ROWID0, &buffer, &size);
553 767 : if (rc != DBRES_OK) return rc;
554 767 : if (!buffer || size != UUID_LEN) {
555 0 : if (buffer) cloudsync_memory_free(buffer);
556 0 : return cloudsync_set_error(data, "Unable to retrieve siteid", DBRES_MISUSE);
557 : }
558 :
559 767 : memcpy(data->site_id, buffer, UUID_LEN);
560 767 : cloudsync_memory_free(buffer);
561 :
562 767 : return DBRES_OK;
563 769 : }
564 :
565 102 : int64_t cloudsync_dbversion (cloudsync_context *data) {
566 102 : return data->db_version;
567 : }
568 :
569 17506 : int cloudsync_bumpseq (cloudsync_context *data) {
570 17506 : int value = data->seq;
571 17506 : data->seq += 1;
572 17506 : return value;
573 : }
574 :
575 821 : void cloudsync_update_schema_hash (cloudsync_context *data) {
576 821 : database_update_schema_hash(data, &data->schema_hash);
577 821 : }
578 :
579 158557 : void *cloudsync_db (cloudsync_context *data) {
580 158557 : return data->db;
581 : }
582 :
583 1590 : int cloudsync_add_dbvms (cloudsync_context *data) {
584 : DEBUG_DBFUNCTION("cloudsync_add_stmts");
585 :
586 1590 : if (data->data_version_stmt == NULL) {
587 767 : int rc = databasevm_prepare(data, SQL_DATA_VERSION, (void **)&data->data_version_stmt, DBFLAG_PERSISTENT);
588 : DEBUG_STMT("data_version_stmt %p", data->data_version_stmt);
589 767 : if (rc != DBRES_OK) return rc;
590 : DEBUG_SQL("data_version_stmt: %s", SQL_DATA_VERSION);
591 767 : }
592 :
593 1590 : if (data->schema_version_stmt == NULL) {
594 767 : int rc = databasevm_prepare(data, SQL_SCHEMA_VERSION, (void **)&data->schema_version_stmt, DBFLAG_PERSISTENT);
595 : DEBUG_STMT("schema_version_stmt %p", data->schema_version_stmt);
596 767 : if (rc != DBRES_OK) return rc;
597 : DEBUG_SQL("schema_version_stmt: %s", SQL_SCHEMA_VERSION);
598 767 : }
599 :
600 1590 : if (data->getset_siteid_stmt == NULL) {
601 : // get and set index of the site_id
602 : // in SQLite, we can’t directly combine an INSERT and a SELECT to both insert a row and return an identifier (rowid) in a single statement,
603 : // however, we can use a workaround by leveraging the INSERT statement with ON CONFLICT DO UPDATE and then combining it with RETURNING rowid
604 767 : int rc = databasevm_prepare(data, SQL_SITEID_GETSET_ROWID_BY_SITEID, (void **)&data->getset_siteid_stmt, DBFLAG_PERSISTENT);
605 : DEBUG_STMT("getset_siteid_stmt %p", data->getset_siteid_stmt);
606 767 : if (rc != DBRES_OK) return rc;
607 : DEBUG_SQL("getset_siteid_stmt: %s", SQL_SITEID_GETSET_ROWID_BY_SITEID);
608 767 : }
609 :
610 1590 : return cloudsync_dbversion_rebuild(data);
611 1590 : }
612 :
613 1258 : int cloudsync_set_error (cloudsync_context *data, const char *err_user, int err_code) {
614 : // force err_code to be something different than OK
615 1258 : if (err_code == DBRES_OK) err_code = database_errcode(data);
616 1258 : if (err_code == DBRES_OK) err_code = DBRES_ERROR;
617 :
618 : // compute a meaningful error message
619 1258 : if (err_user == NULL) {
620 135 : snprintf(data->errmsg, sizeof(data->errmsg), "%s", database_errmsg(data));
621 135 : } else {
622 1123 : const char *db_error = database_errmsg(data);
623 : char db_error_copy[sizeof(data->errmsg)];
624 1123 : int rc = database_errcode(data);
625 1123 : if (rc == DBRES_OK) {
626 : // no database error behind this one, so no SQLSTATE either
627 562 : snprintf(data->errmsg, sizeof(data->errmsg), "%s", err_user);
628 562 : data->sqlstate = 0;
629 562 : } else {
630 561 : if (db_error == data->errmsg) {
631 0 : snprintf(db_error_copy, sizeof(db_error_copy), "%s", db_error);
632 0 : db_error = db_error_copy;
633 0 : }
634 561 : snprintf(data->errmsg, sizeof(data->errmsg), "%s (%s)", err_user, db_error);
635 : }
636 : }
637 :
638 1258 : data->errcode = err_code;
639 1258 : return err_code;
640 : }
641 :
642 135 : int cloudsync_set_dberror (cloudsync_context *data) {
643 135 : return cloudsync_set_error(data, NULL, DBRES_OK);
644 : }
645 :
646 1349 : const char *cloudsync_errmsg (cloudsync_context *data) {
647 1349 : return data->errmsg;
648 : }
649 :
650 39 : void cloudsync_apply_stats_reset (cloudsync_context *data) {
651 39 : if (data) data->apply_rows = 0;
652 39 : }
653 :
654 : // Saturating: only a receive drain resets it, so on the direct-SQL apply path it
655 : // accumulates for the life of the connection and signed overflow would be undefined.
656 1914 : static void cloudsync_apply_stats_add (cloudsync_context *data, int rows) {
657 1914 : if (rows > 0) data->apply_rows = (data->apply_rows > INT_MAX - rows) ? INT_MAX : data->apply_rows + rows;
658 1914 : }
659 :
660 48 : void cloudsync_receive_stream_reset (cloudsync_context *data) {
661 48 : if (data) data->stream_values_count = 0;
662 48 : }
663 :
664 39 : static cloudsync_stream_value *cloudsync_stream_value_find (cloudsync_context *data, const char *value_id) {
665 39 : for (int i = 0; i < data->stream_values_count; i++) {
666 23 : if (strcmp(data->stream_values[i].value_id, value_id) == 0) return &data->stream_values[i];
667 0 : }
668 16 : return NULL;
669 39 : }
670 :
671 : // Records whether the current receive stream has applied value_id or still has it pending.
672 18 : static int cloudsync_stream_value_track (cloudsync_context *data, const char *value_id, bool applied) {
673 18 : cloudsync_stream_value *v = cloudsync_stream_value_find(data, value_id);
674 18 : if (!v) {
675 8 : if (data->stream_values_count == data->stream_values_cap) {
676 6 : int cap = data->stream_values_cap ? data->stream_values_cap * 2 : 8;
677 6 : cloudsync_stream_value *values = cloudsync_memory_realloc(data->stream_values, (uint64_t)cap * sizeof(*values));
678 6 : if (!values) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: out of memory", DBRES_NOMEM);
679 6 : data->stream_values = values;
680 6 : data->stream_values_cap = cap;
681 6 : }
682 8 : v = &data->stream_values[data->stream_values_count++];
683 8 : snprintf(v->value_id, sizeof(v->value_id), "%s", value_id);
684 8 : }
685 18 : v->applied = applied;
686 18 : return DBRES_OK;
687 18 : }
688 :
689 39 : int cloudsync_apply_rows_count (cloudsync_context *data) {
690 39 : return (data) ? data->apply_rows : 0;
691 : }
692 :
693 126 : int cloudsync_errcode (cloudsync_context *data) {
694 126 : return data->errcode;
695 : }
696 :
697 7575 : void cloudsync_reset_error (cloudsync_context *data) {
698 7575 : data->errmsg[0] = 0;
699 7575 : data->errcode = DBRES_OK;
700 7575 : data->sqlstate = 0;
701 7575 : }
702 :
703 614 : void cloudsync_set_sqlstate (cloudsync_context *data, int sqlstate) {
704 614 : if (data) data->sqlstate = sqlstate;
705 614 : }
706 :
707 628 : int cloudsync_sqlstate (cloudsync_context *data) {
708 628 : return (data) ? data->sqlstate : 0;
709 : }
710 :
711 173 : void *cloudsync_auxdata (cloudsync_context *data) {
712 173 : return data->aux_data;
713 : }
714 :
715 37 : void cloudsync_set_auxdata (cloudsync_context *data, void *xdata) {
716 37 : data->aux_data = xdata;
717 37 : }
718 :
719 6 : void cloudsync_set_schema (cloudsync_context *data, const char *schema) {
720 6 : if (data->current_schema && schema && strcmp(data->current_schema, schema) == 0) return;
721 5 : if (data->current_schema) cloudsync_memory_free(data->current_schema);
722 5 : data->current_schema = NULL;
723 5 : if (schema) data->current_schema = cloudsync_string_dup_lowercase(schema);
724 6 : }
725 :
726 3979 : const char *cloudsync_schema (cloudsync_context *data) {
727 3979 : return data->current_schema;
728 : }
729 :
730 4 : const char *cloudsync_table_schema (cloudsync_context *data, const char *table_name) {
731 4 : cloudsync_table_context *table = table_lookup(data, table_name);
732 4 : if (!table) return NULL;
733 :
734 1 : return table->schema;
735 4 : }
736 :
737 : // MARK: - Table Utils -
738 :
739 69 : void table_pknames_free (char **names, int nrows) {
740 69 : if (!names) return;
741 132 : for (int i = 0; i < nrows; ++i) {cloudsync_memory_free(names[i]);}
742 46 : cloudsync_memory_free(names);
743 69 : }
744 :
745 824 : char *table_build_mergedelete_sql (cloudsync_table_context *table) {
746 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
747 : if (table->rowid_only) {
748 : char *sql = memory_mprintf(SQL_DELETE_ROW_BY_ROWID, table->name);
749 : return sql;
750 : }
751 : #endif
752 :
753 824 : return sql_build_delete_by_pk(table->context, table->name, table->schema);
754 : }
755 :
756 2718 : char *table_build_mergeinsert_sql (cloudsync_table_context *table, const char *colname) {
757 2718 : char *sql = NULL;
758 :
759 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
760 : if (table->rowid_only) {
761 : if (colname == NULL) {
762 : // INSERT OR IGNORE INTO customers (first_name,last_name) VALUES (?,?);
763 : sql = memory_mprintf(SQL_INSERT_ROWID_IGNORE, table->name);
764 : } else {
765 : // INSERT INTO customers (first_name,last_name,age) VALUES (?,?,?) ON CONFLICT DO UPDATE SET age=?;
766 : sql = memory_mprintf(SQL_UPSERT_ROWID_AND_COL_BY_ROWID, table->name, colname, colname);
767 : }
768 : return sql;
769 : }
770 : #endif
771 :
772 2718 : if (colname == NULL) {
773 : // is sentinel insert
774 824 : sql = sql_build_insert_pk_ignore(table->context, table->name, table->schema);
775 824 : } else {
776 1894 : sql = sql_build_upsert_pk_and_col(table->context, table->name, colname, table->schema);
777 : }
778 2718 : return sql;
779 : }
780 :
781 1893 : char *table_build_value_sql (cloudsync_table_context *table, const char *colname) {
782 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
783 : if (table->rowid_only) {
784 : char *colnamequote = "\"";
785 : char *sql = memory_mprintf(SQL_SELECT_COLS_BY_ROWID_FMT, colnamequote, colname, colnamequote, table->name);
786 : return sql;
787 : }
788 : #endif
789 :
790 : // SELECT age FROM customers WHERE first_name=? AND last_name=?;
791 1893 : return sql_build_select_cols_by_pk(table->context, table->name, colname, table->schema);
792 : }
793 :
794 824 : cloudsync_table_context *table_create (cloudsync_context *data, const char *name, table_algo algo) {
795 : DEBUG_DBFUNCTION("table_create %s", name);
796 :
797 824 : cloudsync_table_context *table = (cloudsync_table_context *)cloudsync_memory_zeroalloc(sizeof(cloudsync_table_context));
798 824 : if (!table) return NULL;
799 :
800 824 : table->context = data;
801 824 : table->algo = algo;
802 824 : table->name = cloudsync_string_dup_lowercase(name);
803 :
804 : // Detect schema from metadata table location. If metadata table doesn't
805 : // exist yet (during initialization), fall back to cloudsync_schema() which
806 : // returns the explicitly set schema or current_schema().
807 824 : table->schema = database_table_schema(name);
808 824 : if (!table->schema) {
809 824 : const char *fallback_schema = cloudsync_schema(data);
810 824 : if (fallback_schema) {
811 0 : table->schema = cloudsync_string_dup(fallback_schema);
812 0 : }
813 824 : }
814 :
815 824 : if (!table->name) {
816 0 : cloudsync_memory_free(table);
817 0 : return NULL;
818 : }
819 824 : table->meta_ref = database_build_meta_ref(table->schema, table->name);
820 824 : table->base_ref = database_build_base_ref(table->schema, table->name);
821 824 : table->enabled = true;
822 :
823 824 : return table;
824 824 : }
825 :
826 824 : void table_free (cloudsync_table_context *table) {
827 : DEBUG_DBFUNCTION("table_free %s", (table) ? (table->name) : "NULL");
828 824 : if (!table) return;
829 :
830 824 : if (table->col_name) {
831 2677 : for (int i=0; i<table->ncols; ++i) {
832 1893 : cloudsync_memory_free(table->col_name[i]);
833 1893 : }
834 784 : cloudsync_memory_free(table->col_name);
835 784 : }
836 824 : if (table->col_merge_stmt) {
837 2677 : for (int i=0; i<table->ncols; ++i) {
838 1893 : databasevm_finalize(table->col_merge_stmt[i]);
839 1893 : }
840 784 : cloudsync_memory_free(table->col_merge_stmt);
841 784 : }
842 824 : if (table->col_update_stmt) {
843 2677 : for (int i=0; i<table->ncols; ++i) {
844 1893 : if (table->col_update_stmt[i]) databasevm_finalize(table->col_update_stmt[i]);
845 1893 : }
846 784 : cloudsync_memory_free(table->col_update_stmt);
847 784 : }
848 824 : if (table->col_value_stmt) {
849 2677 : for (int i=0; i<table->ncols; ++i) {
850 1893 : databasevm_finalize(table->col_value_stmt[i]);
851 1893 : }
852 784 : cloudsync_memory_free(table->col_value_stmt);
853 784 : }
854 824 : if (table->col_id) {
855 784 : cloudsync_memory_free(table->col_id);
856 784 : }
857 824 : if (table->col_algo) {
858 784 : cloudsync_memory_free(table->col_algo);
859 784 : }
860 824 : if (table->col_delimiter) {
861 2677 : for (int i=0; i<table->ncols; ++i) {
862 1893 : if (table->col_delimiter[i]) cloudsync_memory_free(table->col_delimiter[i]);
863 1893 : }
864 784 : cloudsync_memory_free(table->col_delimiter);
865 784 : }
866 :
867 824 : if (table->block_value_read_stmt) databasevm_finalize(table->block_value_read_stmt);
868 824 : if (table->block_value_write_stmt) databasevm_finalize(table->block_value_write_stmt);
869 824 : if (table->block_value_delete_stmt) databasevm_finalize(table->block_value_delete_stmt);
870 824 : if (table->block_list_stmt) databasevm_finalize(table->block_list_stmt);
871 824 : if (table->blocks_ref) cloudsync_memory_free(table->blocks_ref);
872 :
873 824 : if (table->name) cloudsync_memory_free(table->name);
874 824 : if (table->schema) cloudsync_memory_free(table->schema);
875 824 : if (table->meta_ref) cloudsync_memory_free(table->meta_ref);
876 824 : if (table->base_ref) cloudsync_memory_free(table->base_ref);
877 824 : if (table->pk_name) table_pknames_free(table->pk_name, table->npks);
878 824 : if (table->meta_pkexists_stmt) databasevm_finalize(table->meta_pkexists_stmt);
879 824 : if (table->meta_sentinel_update_stmt) databasevm_finalize(table->meta_sentinel_update_stmt);
880 824 : if (table->meta_sentinel_insert_stmt) databasevm_finalize(table->meta_sentinel_insert_stmt);
881 824 : if (table->meta_row_insert_update_stmt) databasevm_finalize(table->meta_row_insert_update_stmt);
882 824 : if (table->meta_row_drop_stmt) databasevm_finalize(table->meta_row_drop_stmt);
883 824 : if (table->meta_update_move_stmt) databasevm_finalize(table->meta_update_move_stmt);
884 824 : if (table->meta_local_cl_stmt) databasevm_finalize(table->meta_local_cl_stmt);
885 824 : if (table->meta_winner_clock_stmt) databasevm_finalize(table->meta_winner_clock_stmt);
886 824 : if (table->meta_merge_delete_drop) databasevm_finalize(table->meta_merge_delete_drop);
887 824 : if (table->meta_zero_clock_stmt) databasevm_finalize(table->meta_zero_clock_stmt);
888 824 : if (table->meta_col_version_stmt) databasevm_finalize(table->meta_col_version_stmt);
889 824 : if (table->meta_site_id_stmt) databasevm_finalize(table->meta_site_id_stmt);
890 :
891 824 : if (table->real_col_values_stmt) databasevm_finalize(table->real_col_values_stmt);
892 824 : if (table->real_merge_delete_stmt) databasevm_finalize(table->real_merge_delete_stmt);
893 824 : if (table->real_merge_sentinel_stmt) databasevm_finalize(table->real_merge_sentinel_stmt);
894 :
895 824 : cloudsync_memory_free(table);
896 824 : }
897 :
898 824 : int table_add_stmts (cloudsync_table_context *table, int ncols) {
899 824 : int rc = DBRES_OK;
900 824 : char *sql = NULL;
901 824 : cloudsync_context *data = table->context;
902 :
903 : // META TABLE statements
904 :
905 : // CREATE TABLE IF NOT EXISTS \"%w_cloudsync\" (pk BLOB NOT NULL, col_name TEXT NOT NULL, col_version INTEGER, db_version INTEGER, site_id INTEGER DEFAULT 0, seq INTEGER, PRIMARY KEY (pk, col_name));
906 :
907 : // precompile the pk exists statement
908 : // we do not need an index on the pk column because it is already covered by the fact that it is part of the prikeys
909 : // EXPLAIN QUERY PLAN reports: SEARCH table_name USING PRIMARY KEY (pk=?)
910 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_ROW_EXISTS_BY_PK, table->meta_ref);
911 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
912 : DEBUG_SQL("meta_pkexists_stmt: %s", sql);
913 :
914 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_pkexists_stmt, DBFLAG_PERSISTENT);
915 824 : cloudsync_memory_free(sql);
916 824 : if (rc != DBRES_OK) goto cleanup;
917 :
918 : // precompile the update local sentinel statement
919 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPDATE_COL_BUMP_VERSION, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
920 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
921 : DEBUG_SQL("meta_sentinel_update_stmt: %s", sql);
922 :
923 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_sentinel_update_stmt, DBFLAG_PERSISTENT);
924 824 : cloudsync_memory_free(sql);
925 824 : if (rc != DBRES_OK) goto cleanup;
926 :
927 : // precompile the insert local sentinel statement
928 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_COL_INIT_OR_BUMP_VERSION, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE, table->meta_ref, table->meta_ref, table->meta_ref);
929 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
930 : DEBUG_SQL("meta_sentinel_insert_stmt: %s", sql);
931 :
932 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_sentinel_insert_stmt, DBFLAG_PERSISTENT);
933 824 : cloudsync_memory_free(sql);
934 824 : if (rc != DBRES_OK) goto cleanup;
935 :
936 : // precompile the insert/update local row statement
937 1648 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION,
938 824 : table->meta_ref, table->meta_ref,
939 824 : table->meta_ref, table->meta_ref);
940 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
941 : DEBUG_SQL("meta_row_insert_update_stmt: %s", sql);
942 :
943 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_row_insert_update_stmt, DBFLAG_PERSISTENT);
944 824 : cloudsync_memory_free(sql);
945 824 : if (rc != DBRES_OK) goto cleanup;
946 :
947 : // precompile the delete rows from meta
948 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
949 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
950 : DEBUG_SQL("meta_row_drop_stmt: %s", sql);
951 :
952 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_row_drop_stmt, DBFLAG_PERSISTENT);
953 824 : cloudsync_memory_free(sql);
954 824 : if (rc != DBRES_OK) goto cleanup;
955 :
956 : // precompile the update rows from meta when pk changes
957 : // see https://github.com/sqliteai/sqlite-sync/blob/main/docs/PriKey.md for more details
958 824 : sql = sql_build_rekey_pk_and_reset_version_except_col(data, table->name, CLOUDSYNC_TOMBSTONE_VALUE);
959 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
960 : DEBUG_SQL("meta_update_move_stmt: %s", sql);
961 :
962 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_update_move_stmt, DBFLAG_PERSISTENT);
963 824 : cloudsync_memory_free(sql);
964 824 : if (rc != DBRES_OK) goto cleanup;
965 :
966 : // local cl
967 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_GET_COL_VERSION_OR_ROW_EXISTS, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE, table->meta_ref);
968 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
969 : DEBUG_SQL("meta_local_cl_stmt: %s", sql);
970 :
971 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_local_cl_stmt, DBFLAG_PERSISTENT);
972 824 : cloudsync_memory_free(sql);
973 824 : if (rc != DBRES_OK) goto cleanup;
974 :
975 : // rowid of the last inserted/updated row in the meta table
976 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_INSERT_RETURN_CHANGE_ID, table->meta_ref);
977 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
978 : DEBUG_SQL("meta_winner_clock_stmt: %s", sql);
979 :
980 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_winner_clock_stmt, DBFLAG_PERSISTENT);
981 824 : cloudsync_memory_free(sql);
982 824 : if (rc != DBRES_OK) goto cleanup;
983 :
984 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
985 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
986 : DEBUG_SQL("meta_merge_delete_drop: %s", sql);
987 :
988 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_merge_delete_drop, DBFLAG_PERSISTENT);
989 824 : cloudsync_memory_free(sql);
990 824 : if (rc != DBRES_OK) goto cleanup;
991 :
992 : // zero clock
993 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_TOMBSTONE_PK_EXCEPT_COL, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
994 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
995 : DEBUG_SQL("meta_zero_clock_stmt: %s", sql);
996 :
997 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_zero_clock_stmt, DBFLAG_PERSISTENT);
998 824 : cloudsync_memory_free(sql);
999 824 : if (rc != DBRES_OK) goto cleanup;
1000 :
1001 : // col_version
1002 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_COL_VERSION_BY_PK_COL, table->meta_ref);
1003 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
1004 : DEBUG_SQL("meta_col_version_stmt: %s", sql);
1005 :
1006 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_col_version_stmt, DBFLAG_PERSISTENT);
1007 824 : cloudsync_memory_free(sql);
1008 824 : if (rc != DBRES_OK) goto cleanup;
1009 :
1010 : // site_id
1011 824 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_SITE_ID_BY_PK_COL, table->meta_ref);
1012 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
1013 : DEBUG_SQL("meta_site_id_stmt: %s", sql);
1014 :
1015 824 : rc = databasevm_prepare(data, sql, (void **)&table->meta_site_id_stmt, DBFLAG_PERSISTENT);
1016 824 : cloudsync_memory_free(sql);
1017 824 : if (rc != DBRES_OK) goto cleanup;
1018 :
1019 : // REAL TABLE statements
1020 :
1021 : // precompile the get column value statement
1022 824 : if (ncols > 0) {
1023 784 : sql = sql_build_select_nonpk_by_pk(data, table->name, table->schema);
1024 784 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
1025 : DEBUG_SQL("real_col_values_stmt: %s", sql);
1026 :
1027 784 : rc = databasevm_prepare(data, sql, (void **)&table->real_col_values_stmt, DBFLAG_PERSISTENT);
1028 784 : cloudsync_memory_free(sql);
1029 784 : if (rc != DBRES_OK) goto cleanup;
1030 784 : }
1031 :
1032 824 : sql = table_build_mergedelete_sql(table);
1033 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
1034 : DEBUG_SQL("real_merge_delete: %s", sql);
1035 :
1036 824 : rc = databasevm_prepare(data, sql, (void **)&table->real_merge_delete_stmt, DBFLAG_PERSISTENT);
1037 824 : cloudsync_memory_free(sql);
1038 824 : if (rc != DBRES_OK) goto cleanup;
1039 :
1040 824 : sql = table_build_mergeinsert_sql(table, NULL);
1041 824 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
1042 : DEBUG_SQL("real_merge_sentinel: %s", sql);
1043 :
1044 824 : rc = databasevm_prepare(data, sql, (void **)&table->real_merge_sentinel_stmt, DBFLAG_PERSISTENT);
1045 824 : cloudsync_memory_free(sql);
1046 824 : if (rc != DBRES_OK) goto cleanup;
1047 :
1048 : cleanup:
1049 824 : if (rc != DBRES_OK) DEBUG_ALWAYS("table_add_stmts error: %d %s\n", rc, database_errmsg(data));
1050 824 : return rc;
1051 : }
1052 :
1053 228847 : cloudsync_table_context *table_lookup (cloudsync_context *data, const char *table_name) {
1054 : DEBUG_DBFUNCTION("table_lookup %s", table_name);
1055 :
1056 228847 : if (table_name) {
1057 230473 : for (int i=0; i<data->tables_count; ++i) {
1058 228818 : if ((strcasecmp(data->tables[i]->name, table_name) == 0)) return data->tables[i];
1059 1626 : }
1060 1655 : }
1061 :
1062 1655 : return NULL;
1063 228847 : }
1064 :
1065 189912 : void *table_column_lookup (cloudsync_table_context *table, const char *col_name, bool is_merge, int *index) {
1066 : DEBUG_DBFUNCTION("table_column_lookup %s", col_name);
1067 :
1068 339682 : for (int i=0; i<table->ncols; ++i) {
1069 339682 : if (strcasecmp(table->col_name[i], col_name) == 0) {
1070 189912 : if (index) *index = i;
1071 189912 : return (is_merge) ? table->col_merge_stmt[i] : table->col_value_stmt[i];
1072 : }
1073 149770 : }
1074 :
1075 0 : if (index) *index = -1;
1076 0 : return NULL;
1077 189912 : }
1078 :
1079 823 : int table_remove (cloudsync_context *data, cloudsync_table_context *table) {
1080 823 : const char *table_name = table->name;
1081 : DEBUG_DBFUNCTION("table_remove %s", table_name);
1082 :
1083 873 : for (int i = 0; i < data->tables_count; ++i) {
1084 873 : cloudsync_table_context *t = data->tables[i];
1085 :
1086 : // pointer compare is fastest but fallback to strcasecmp if not same pointer
1087 873 : if ((t == table) || ((strcasecmp(t->name, table_name) == 0))) {
1088 823 : int last = data->tables_count - 1;
1089 823 : data->tables[i] = data->tables[last]; // move last into the hole (keeps array dense)
1090 823 : data->tables[last] = NULL; // NULLify tail (as an extra security measure)
1091 823 : data->tables_count--;
1092 823 : return data->tables_count;
1093 : }
1094 50 : }
1095 :
1096 0 : return -1;
1097 823 : }
1098 :
1099 1894 : int table_add_to_context_cb (void *xdata, int ncols, char **values, char **names) {
1100 1894 : cloudsync_table_context *table = (cloudsync_table_context *)xdata;
1101 1894 : cloudsync_context *data = table->context;
1102 :
1103 1894 : int index = table->ncols;
1104 3787 : for (int i=0; i<ncols; i+=2) {
1105 1894 : const char *name = values[i];
1106 1894 : int cid = (int)strtol(values[i+1], NULL, 0);
1107 :
1108 1894 : table->col_id[index] = cid;
1109 1894 : table->col_name[index] = cloudsync_string_dup_lowercase(name);
1110 1894 : if (!table->col_name[index]) goto error;
1111 :
1112 1894 : char *sql = table_build_mergeinsert_sql(table, name);
1113 1894 : if (!sql) goto error;
1114 : DEBUG_SQL("col_merge_stmt[%d]: %s", index, sql);
1115 :
1116 1894 : int rc = databasevm_prepare(data, sql, (void **)&table->col_merge_stmt[index], DBFLAG_PERSISTENT);
1117 1894 : cloudsync_memory_free(sql);
1118 1894 : if (rc != DBRES_OK) goto error;
1119 1893 : if (!table->col_merge_stmt[index]) goto error;
1120 :
1121 1893 : sql = table_build_value_sql(table, name);
1122 1893 : if (!sql) goto error;
1123 : DEBUG_SQL("col_value_stmt[%d]: %s", index, sql);
1124 :
1125 1893 : rc = databasevm_prepare(data, sql, (void **)&table->col_value_stmt[index], DBFLAG_PERSISTENT);
1126 1893 : cloudsync_memory_free(sql);
1127 1893 : if (rc != DBRES_OK) goto error;
1128 1893 : if (!table->col_value_stmt[index]) goto error;
1129 1893 : }
1130 1893 : table->ncols += 1;
1131 :
1132 1893 : return 0;
1133 :
1134 : error:
1135 : // clean up partially-initialized entry at index
1136 1 : if (table->col_name[index]) {cloudsync_memory_free(table->col_name[index]); table->col_name[index] = NULL;}
1137 1 : if (table->col_merge_stmt[index]) {databasevm_finalize(table->col_merge_stmt[index]); table->col_merge_stmt[index] = NULL;}
1138 1 : if (table->col_value_stmt[index]) {databasevm_finalize(table->col_value_stmt[index]); table->col_value_stmt[index] = NULL;}
1139 1 : return 1;
1140 1894 : }
1141 :
1142 824 : bool table_ensure_capacity (cloudsync_context *data) {
1143 824 : if (data->tables_count < data->tables_cap) return true;
1144 :
1145 0 : int new_cap = data->tables_cap ? data->tables_cap * 2 : CLOUDSYNC_INIT_NTABLES;
1146 0 : size_t bytes = (size_t)new_cap * sizeof(*data->tables);
1147 0 : void *p = cloudsync_memory_realloc(data->tables, bytes);
1148 0 : if (!p) return false;
1149 :
1150 0 : data->tables = (cloudsync_table_context **)p;
1151 0 : data->tables_cap = new_cap;
1152 0 : return true;
1153 824 : }
1154 :
1155 832 : bool table_add_to_context (cloudsync_context *data, table_algo algo, const char *table_name) {
1156 : DEBUG_DBFUNCTION("cloudsync_context_add_table %s", table_name);
1157 :
1158 : // Check if table already initialized in this connection's context.
1159 : // Note: This prevents same-connection duplicate initialization.
1160 : // SQLite clients cannot distinguish schemas, so having 'public.users'
1161 : // and 'auth.users' would cause sync ambiguity. Users should avoid
1162 : // initializing tables with the same name in different schemas.
1163 : // If two concurrent connections initialize tables with the same name
1164 : // in different schemas, the behavior is undefined.
1165 832 : cloudsync_table_context *table = table_lookup(data, table_name);
1166 832 : if (table) return true;
1167 :
1168 : // check for space availability
1169 824 : if (!table_ensure_capacity(data)) return false;
1170 :
1171 : // setup a new table
1172 824 : table = table_create(data, table_name, algo);
1173 824 : if (!table) return false;
1174 :
1175 : // fill remaining metadata in the table
1176 824 : int count = database_count_pk(data, table_name, false, table->schema);
1177 824 : if (count < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
1178 824 : table->npks = count;
1179 824 : if (table->npks == 0) {
1180 : #if CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
1181 0 : goto abort_add_table;
1182 : #else
1183 : table->rowid_only = true;
1184 : table->npks = 1; // rowid
1185 : #endif
1186 : }
1187 :
1188 824 : int ncols = database_count_nonpk(data, table_name, table->schema);
1189 824 : if (ncols < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
1190 824 : int rc = table_add_stmts(table, ncols);
1191 824 : if (rc != DBRES_OK) goto abort_add_table;
1192 :
1193 : // a table with only pk(s) is totally legal
1194 824 : if (ncols > 0) {
1195 784 : table->col_name = (char **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(char *) * ncols));
1196 784 : if (!table->col_name) goto abort_add_table;
1197 :
1198 784 : table->col_id = (int *)cloudsync_memory_zeroalloc((uint64_t)(sizeof(int) * ncols));
1199 784 : if (!table->col_id) goto abort_add_table;
1200 :
1201 784 : table->col_merge_stmt = (dbvm_t **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(void *) * ncols));
1202 784 : if (!table->col_merge_stmt) goto abort_add_table;
1203 :
1204 784 : table->col_update_stmt = (dbvm_t **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(void *) * ncols));
1205 784 : if (!table->col_update_stmt) goto abort_add_table;
1206 :
1207 784 : table->col_value_stmt = (dbvm_t **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(void *) * ncols));
1208 784 : if (!table->col_value_stmt) goto abort_add_table;
1209 :
1210 784 : table->col_algo = (col_algo_t *)cloudsync_memory_zeroalloc((uint64_t)(sizeof(col_algo_t) * ncols));
1211 784 : if (!table->col_algo) goto abort_add_table;
1212 :
1213 784 : table->col_delimiter = (char **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(char *) * ncols));
1214 784 : if (!table->col_delimiter) goto abort_add_table;
1215 :
1216 : // Pass empty string when schema is NULL; SQL will fall back to current_schema()
1217 784 : const char *schema = table->schema ? table->schema : "";
1218 1568 : char *sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_LIST_NONPK_NAME_CID,
1219 784 : table_name, schema, table_name, schema);
1220 784 : if (!sql) goto abort_add_table;
1221 784 : rc = database_exec_callback(data, sql, table_add_to_context_cb, (void *)table);
1222 784 : cloudsync_memory_free(sql);
1223 784 : if (rc == DBRES_ABORT) goto abort_add_table;
1224 783 : }
1225 :
1226 : // append newly created table
1227 823 : data->tables[data->tables_count++] = table;
1228 823 : return true;
1229 :
1230 : abort_add_table:
1231 1 : table_free(table);
1232 1 : return false;
1233 832 : }
1234 :
1235 0 : dbvm_t *cloudsync_colvalue_stmt (cloudsync_context *data, const char *tbl_name, bool *persistent) {
1236 0 : dbvm_t *vm = NULL;
1237 0 : *persistent = false;
1238 :
1239 0 : cloudsync_table_context *table = table_lookup(data, tbl_name);
1240 0 : if (table) {
1241 0 : char *col_name = NULL;
1242 0 : if (table->ncols > 0) {
1243 0 : col_name = table->col_name[0];
1244 : // retrieve col_value precompiled statement
1245 0 : vm = table_column_lookup(table, col_name, false, NULL);
1246 0 : *persistent = true;
1247 0 : } else {
1248 0 : char *sql = table_build_value_sql(table, "*");
1249 0 : databasevm_prepare(data, sql, (void **)&vm, 0);
1250 0 : cloudsync_memory_free(sql);
1251 0 : *persistent = false;
1252 : }
1253 0 : }
1254 :
1255 0 : return vm;
1256 : }
1257 :
1258 9639 : bool table_enabled (cloudsync_table_context *table) {
1259 9639 : return table->enabled;
1260 : }
1261 :
1262 10 : void table_set_enabled (cloudsync_table_context *table, bool value) {
1263 10 : table->enabled = value;
1264 10 : }
1265 :
1266 28300 : int table_count_cols (cloudsync_table_context *table) {
1267 28300 : return table->ncols;
1268 : }
1269 :
1270 10745 : int table_count_pks (cloudsync_table_context *table) {
1271 10745 : return table->npks;
1272 : }
1273 :
1274 39260 : const char *table_colname (cloudsync_table_context *table, int index) {
1275 39260 : return table->col_name[index];
1276 : }
1277 :
1278 6012 : bool table_pk_exists (cloudsync_table_context *table, const char *value, size_t len) {
1279 : // check if a row with the same primary key already exists
1280 : // if so, this means the row might have been previously deleted (sentinel)
1281 6012 : return (dbvm_count(table->meta_pkexists_stmt, value, len, DBTYPE_BLOB) > 0);
1282 : }
1283 :
1284 0 : char **table_pknames (cloudsync_table_context *table) {
1285 0 : return table->pk_name;
1286 : }
1287 :
1288 23 : void table_set_pknames (cloudsync_table_context *table, char **pknames) {
1289 23 : table_pknames_free(table->pk_name, table->npks);
1290 23 : table->pk_name = pknames;
1291 23 : }
1292 :
1293 58062 : bool table_algo_isgos (cloudsync_table_context *table) {
1294 58062 : return (table->algo == table_algo_crdt_gos);
1295 : }
1296 :
1297 0 : const char *table_schema (cloudsync_table_context *table) {
1298 0 : return table->schema;
1299 : }
1300 :
1301 : // MARK: - Merge Insert -
1302 :
1303 58062 : int64_t merge_get_local_cl (cloudsync_table_context *table, const char *pk, int pklen) {
1304 58062 : dbvm_t *vm = table->meta_local_cl_stmt;
1305 58062 : int64_t result = -1;
1306 :
1307 58062 : int rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1308 58062 : if (rc != DBRES_OK) goto cleanup;
1309 :
1310 58062 : rc = databasevm_bind_blob(vm, 2, (const void *)pk, pklen);
1311 58062 : if (rc != DBRES_OK) goto cleanup;
1312 :
1313 58062 : rc = databasevm_step(vm);
1314 58062 : if (rc == DBRES_ROW) result = database_column_int(vm, 0);
1315 0 : else if (rc == DBRES_DONE) result = 0;
1316 :
1317 : cleanup:
1318 58062 : if (result == -1) cloudsync_set_dberror(table->context);
1319 58062 : dbvm_reset(vm);
1320 58062 : return result;
1321 : }
1322 :
1323 53556 : int merge_get_col_version (cloudsync_table_context *table, const char *col_name, const char *pk, int pklen, int64_t *version) {
1324 53556 : dbvm_t *vm = table->meta_col_version_stmt;
1325 :
1326 53556 : int rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1327 53556 : if (rc != DBRES_OK) goto cleanup;
1328 :
1329 53556 : rc = databasevm_bind_text(vm, 2, col_name, -1);
1330 53556 : if (rc != DBRES_OK) goto cleanup;
1331 :
1332 53556 : rc = databasevm_step(vm);
1333 82934 : if (rc == DBRES_ROW) {
1334 29378 : *version = database_column_int(vm, 0);
1335 29378 : rc = DBRES_OK;
1336 29378 : }
1337 :
1338 : cleanup:
1339 53556 : if ((rc != DBRES_OK) && (rc != DBRES_DONE)) cloudsync_set_dberror(table->context);
1340 53556 : dbvm_reset(vm);
1341 53556 : return rc;
1342 : }
1343 :
1344 31376 : int merge_set_winner_clock (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pk_len, const char *colname, int64_t col_version, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1345 :
1346 : // get/set site_id
1347 31376 : dbvm_t *vm = data->getset_siteid_stmt;
1348 31376 : int rc = databasevm_bind_blob(vm, 1, (const void *)site_id, site_len);
1349 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1350 :
1351 31376 : rc = databasevm_step(vm);
1352 31376 : if (rc != DBRES_ROW) goto cleanup_merge;
1353 :
1354 31376 : int64_t ord = database_column_int(vm, 0);
1355 31376 : dbvm_reset(vm);
1356 :
1357 31376 : vm = table->meta_winner_clock_stmt;
1358 31376 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pk_len);
1359 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1360 :
1361 31376 : rc = databasevm_bind_text(vm, 2, (colname) ? colname : CLOUDSYNC_TOMBSTONE_VALUE, -1);
1362 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1363 :
1364 31376 : rc = databasevm_bind_int(vm, 3, col_version);
1365 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1366 :
1367 31376 : rc = databasevm_bind_int(vm, 4, db_version);
1368 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1369 :
1370 31376 : rc = databasevm_bind_int(vm, 5, seq);
1371 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1372 :
1373 31376 : rc = databasevm_bind_int(vm, 6, ord);
1374 31376 : if (rc != DBRES_OK) goto cleanup_merge;
1375 :
1376 31376 : rc = databasevm_step(vm);
1377 62752 : if (rc == DBRES_ROW) {
1378 31376 : *rowid = database_column_int(vm, 0);
1379 31376 : rc = DBRES_OK;
1380 31376 : }
1381 :
1382 : cleanup_merge:
1383 31376 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1384 31376 : dbvm_reset(vm);
1385 31376 : return rc;
1386 : }
1387 :
1388 : // MARK: - Deferred column-batch merge functions -
1389 :
1390 23703 : static int merge_pending_add (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, const char *col_name, dbvalue_t *col_value, int64_t col_version, int64_t db_version, const char *site_id, int site_len, int64_t seq) {
1391 23703 : merge_pending_batch *batch = data->pending_batch;
1392 :
1393 : // Store table and PK on first entry
1394 23703 : if (batch->table == NULL) {
1395 9432 : batch->table = table;
1396 9432 : batch->pk = (char *)cloudsync_memory_alloc(pklen);
1397 9432 : if (!batch->pk) return cloudsync_set_error(data, "merge_pending_add: out of memory for pk", DBRES_NOMEM);
1398 9432 : memcpy(batch->pk, pk, pklen);
1399 9432 : batch->pk_len = pklen;
1400 9432 : }
1401 :
1402 : // Ensure capacity
1403 23703 : if (batch->count >= batch->capacity) {
1404 1142 : int new_cap = batch->capacity ? batch->capacity * 2 : 8;
1405 1142 : merge_pending_entry *new_entries = (merge_pending_entry *)cloudsync_memory_realloc(batch->entries, new_cap * sizeof(merge_pending_entry));
1406 1142 : if (!new_entries) return cloudsync_set_error(data, "merge_pending_add: out of memory for entries", DBRES_NOMEM);
1407 1142 : batch->entries = new_entries;
1408 1142 : batch->capacity = new_cap;
1409 1142 : }
1410 :
1411 : // Resolve col_name to a stable pointer from the table context
1412 : // (the incoming col_name may point to VM-owned memory that gets freed on reset)
1413 23703 : int col_idx = -1;
1414 23703 : table_column_lookup(table, col_name, true, &col_idx);
1415 23703 : const char *stable_col_name = (col_idx >= 0) ? table_colname(table, col_idx) : NULL;
1416 23703 : if (!stable_col_name) return cloudsync_set_error(data, "merge_pending_add: column not found in table context", DBRES_ERROR);
1417 :
1418 23703 : merge_pending_entry *e = &batch->entries[batch->count];
1419 23703 : e->col_name = stable_col_name;
1420 23703 : e->col_value = col_value ? (dbvalue_t *)database_value_dup(col_value) : NULL;
1421 23703 : if (col_value && !e->col_value) return DBRES_NOMEM;
1422 23703 : e->col_version = col_version;
1423 23703 : e->db_version = db_version;
1424 23703 : e->site_id_len = (site_len <= (int)sizeof(e->site_id)) ? site_len : (int)sizeof(e->site_id);
1425 23703 : memcpy(e->site_id, site_id, e->site_id_len);
1426 23703 : e->seq = seq;
1427 :
1428 23703 : batch->count++;
1429 23703 : return DBRES_OK;
1430 23703 : }
1431 :
1432 26257 : static void merge_pending_free_entries (merge_pending_batch *batch) {
1433 26257 : if (batch->entries) {
1434 41495 : for (int i = 0; i < batch->count; i++) {
1435 23703 : if (batch->entries[i].col_value) {
1436 23703 : database_value_free(batch->entries[i].col_value);
1437 23703 : batch->entries[i].col_value = NULL;
1438 23703 : }
1439 23703 : }
1440 17792 : }
1441 26257 : if (batch->pk) {
1442 9672 : cloudsync_memory_free(batch->pk);
1443 9672 : batch->pk = NULL;
1444 9672 : }
1445 26257 : batch->table = NULL;
1446 26257 : batch->pk_len = 0;
1447 26257 : batch->cl = 0;
1448 26257 : batch->sentinel_pending = false;
1449 26257 : batch->row_exists = false;
1450 26257 : batch->count = 0;
1451 26257 : }
1452 :
1453 24012 : static int merge_flush_pending (cloudsync_context *data) {
1454 24012 : merge_pending_batch *batch = data->pending_batch;
1455 24012 : if (!batch) return DBRES_OK;
1456 :
1457 24012 : int rc = DBRES_OK;
1458 24012 : bool flush_savepoint = false;
1459 24012 : char error_message[1024] = {0};
1460 24012 : int error_sqlstate = 0;
1461 :
1462 : // Nothing to write — handle sentinel-only case or skip
1463 24012 : if (batch->count == 0 && !(batch->sentinel_pending && batch->table)) {
1464 14340 : goto cleanup;
1465 : }
1466 :
1467 : // Wrap database operations in a savepoint so that on failure (e.g. RLS
1468 : // denial) the rollback properly releases all executor resources (open
1469 : // relations, snapshots, plan cache) acquired during the failed statement.
1470 : // Inside a payload apply's PK-group savepoint the caller rolls back instead.
1471 9672 : if (!batch->group_savepoint) {
1472 19 : rc = database_begin_savepoint(data, "merge_flush");
1473 19 : if (rc != DBRES_OK) goto cleanup;
1474 19 : flush_savepoint = true;
1475 19 : }
1476 :
1477 9672 : if (batch->count == 0) {
1478 : // Sentinel with no winning columns (PK-only row)
1479 149 : dbvm_t *vm = batch->table->real_merge_sentinel_stmt;
1480 149 : rc = pk_decode_prikey(batch->pk, (size_t)batch->pk_len, pk_decode_bind_callback, vm);
1481 149 : if (rc < 0) {
1482 0 : cloudsync_set_dberror(data);
1483 0 : dbvm_reset(vm);
1484 0 : goto cleanup;
1485 : }
1486 149 : SYNCBIT_SET(data);
1487 149 : rc = databasevm_step(vm);
1488 149 : dbvm_reset(vm);
1489 149 : SYNCBIT_RESET(data);
1490 149 : if (rc == DBRES_DONE) rc = DBRES_OK;
1491 149 : if (rc != DBRES_OK) {
1492 0 : cloudsync_set_dberror(data);
1493 0 : goto cleanup;
1494 : }
1495 149 : goto cleanup;
1496 : }
1497 :
1498 : // Check if cached prepared statement can be reused
1499 9523 : cloudsync_table_context *table = batch->table;
1500 9523 : dbvm_t *vm = NULL;
1501 : write_row:;
1502 9524 : bool cache_hit = false;
1503 :
1504 17597 : if (batch->cached_vm &&
1505 8382 : batch->cached_row_exists == batch->row_exists &&
1506 8073 : batch->cached_col_count == batch->count) {
1507 8048 : cache_hit = true;
1508 29428 : for (int i = 0; i < batch->count; i++) {
1509 21406 : if (batch->cached_col_names[i] != batch->entries[i].col_name) {
1510 26 : cache_hit = false;
1511 26 : break;
1512 : }
1513 21380 : }
1514 8048 : }
1515 :
1516 9524 : if (cache_hit) {
1517 8022 : vm = batch->cached_vm;
1518 8022 : dbvm_reset(vm);
1519 8022 : } else {
1520 : // Invalidate old cache
1521 1502 : if (batch->cached_vm) {
1522 360 : databasevm_finalize(batch->cached_vm);
1523 360 : batch->cached_vm = NULL;
1524 360 : }
1525 :
1526 : // Build multi-column SQL
1527 1502 : const char **colnames = (const char **)cloudsync_memory_alloc(batch->count * sizeof(const char *));
1528 1502 : if (!colnames) {
1529 0 : rc = cloudsync_set_error(data, "merge_flush_pending: out of memory", DBRES_NOMEM);
1530 0 : goto cleanup;
1531 : }
1532 3827 : for (int i = 0; i < batch->count; i++) {
1533 2325 : colnames[i] = batch->entries[i].col_name;
1534 2325 : }
1535 :
1536 1502 : char *sql = batch->row_exists
1537 517 : ? sql_build_update_pk_and_multi_cols(data, table->name, colnames, batch->count, table->schema)
1538 985 : : sql_build_upsert_pk_and_multi_cols(data, table->name, colnames, batch->count, table->schema);
1539 1502 : cloudsync_memory_free(colnames);
1540 :
1541 1502 : if (!sql) {
1542 0 : rc = cloudsync_set_error(data, "merge_flush_pending: unable to build multi-column upsert SQL", DBRES_ERROR);
1543 0 : goto cleanup;
1544 : }
1545 :
1546 1502 : rc = databasevm_prepare(data, sql, &vm, 0);
1547 1502 : cloudsync_memory_free(sql);
1548 1502 : if (rc != DBRES_OK) {
1549 0 : rc = cloudsync_set_error(data, "merge_flush_pending: unable to prepare statement", rc);
1550 0 : goto cleanup;
1551 : }
1552 :
1553 : // Update cache
1554 1502 : batch->cached_vm = vm;
1555 1502 : batch->cached_row_exists = batch->row_exists;
1556 1502 : batch->cached_col_count = batch->count;
1557 : // Reallocate cached_col_names if needed
1558 1502 : if (batch->cached_col_count > 0) {
1559 1502 : const char **new_names = (const char **)cloudsync_memory_realloc(
1560 1502 : batch->cached_col_names, batch->count * sizeof(const char *));
1561 1502 : if (!new_names) {
1562 0 : batch->cached_col_count = 0;
1563 0 : rc = DBRES_NOMEM;
1564 0 : goto cleanup;
1565 : }
1566 3827 : for (int i = 0; i < batch->count; i++) {
1567 2325 : new_names[i] = batch->entries[i].col_name;
1568 2325 : }
1569 1502 : batch->cached_col_names = new_names;
1570 1502 : }
1571 : }
1572 :
1573 : // Bind PKs (positions 1..npks)
1574 9524 : int npks = pk_decode_prikey(batch->pk, (size_t)batch->pk_len, pk_decode_bind_callback, vm);
1575 9524 : if (npks < 0) {
1576 0 : cloudsync_set_dberror(data);
1577 0 : dbvm_reset(vm);
1578 0 : rc = DBRES_ERROR;
1579 0 : goto cleanup;
1580 : }
1581 :
1582 : // Bind column values (positions npks+1..npks+count)
1583 33229 : for (int i = 0; i < batch->count; i++) {
1584 23705 : merge_pending_entry *e = &batch->entries[i];
1585 23705 : int bind_idx = npks + 1 + i;
1586 23705 : if (e->col_value) {
1587 23705 : rc = databasevm_bind_value(vm, bind_idx, e->col_value);
1588 23705 : } else {
1589 0 : rc = databasevm_bind_null(vm, bind_idx);
1590 : }
1591 23705 : if (rc != DBRES_OK) {
1592 0 : cloudsync_set_dberror(data);
1593 0 : dbvm_reset(vm);
1594 0 : goto cleanup;
1595 : }
1596 23705 : }
1597 :
1598 : // Execute with SYNCBIT and GOS handling
1599 9524 : if (table->algo == table_algo_crdt_gos) table->enabled = 0;
1600 9524 : SYNCBIT_SET(data);
1601 9524 : rc = databasevm_step(vm);
1602 9524 : bool update_missed = batch->row_exists && rc == DBRES_DONE && databasevm_changes(vm) == 0;
1603 9524 : dbvm_reset(vm);
1604 9524 : SYNCBIT_RESET(data);
1605 9524 : if (table->algo == table_algo_crdt_gos) table->enabled = 1;
1606 :
1607 9524 : if (rc != DBRES_DONE) {
1608 13 : cloudsync_set_dberror(data);
1609 13 : goto cleanup;
1610 : }
1611 9511 : if (update_missed) {
1612 : // The row is not there to update — gone, or hidden by a policy's USING clause,
1613 : // which an UPDATE skips silently. Write it through the upsert instead, which
1614 : // inserts a missing row or reports the policy, rather than record winner clocks
1615 : // for a write that never happened.
1616 1 : batch->row_exists = false;
1617 1 : goto write_row;
1618 : }
1619 9510 : rc = DBRES_OK;
1620 :
1621 : // Call merge_set_winner_clock for each buffered entry
1622 9510 : int64_t rowid = 0;
1623 33192 : for (int i = 0; i < batch->count; i++) {
1624 23682 : merge_pending_entry *e = &batch->entries[i];
1625 47364 : int clock_rc = merge_set_winner_clock(data, table, batch->pk, batch->pk_len,
1626 23682 : e->col_name, e->col_version, e->db_version,
1627 23682 : (const char *)e->site_id, e->site_id_len,
1628 23682 : e->seq, &rowid);
1629 23682 : if (clock_rc != DBRES_OK) {
1630 0 : rc = clock_rc;
1631 0 : goto cleanup;
1632 : }
1633 33192 : }
1634 :
1635 : cleanup:
1636 24012 : if (rc != DBRES_OK) {
1637 13 : snprintf(error_message, sizeof(error_message), "%s", cloudsync_errmsg(data));
1638 13 : error_sqlstate = cloudsync_sqlstate(data);
1639 13 : }
1640 24012 : merge_pending_free_entries(batch);
1641 24012 : if (flush_savepoint) {
1642 19 : if (rc == DBRES_OK) {
1643 18 : rc = database_commit_savepoint(data, "merge_flush");
1644 : // Snapshot here too: arriving with rc OK leaves error_message empty, and a
1645 : // commit that fails (a deadlock or serialization failure, say) sets a real
1646 : // message that the generic fallback below would otherwise replace. The
1647 : // rollback runs after, and touches the error state itself.
1648 18 : if (rc != DBRES_OK) {
1649 0 : snprintf(error_message, sizeof(error_message), "%s", cloudsync_errmsg(data));
1650 0 : error_sqlstate = cloudsync_sqlstate(data);
1651 0 : }
1652 18 : }
1653 19 : if (rc != DBRES_OK) database_rollback_savepoint(data, "merge_flush");
1654 19 : }
1655 24012 : if (rc != DBRES_OK && flush_savepoint) {
1656 : // the rollback cleared the error state: restore the failure's own message and SQLSTATE
1657 1 : cloudsync_reset_error(data);
1658 1 : cloudsync_set_error(data, error_message[0] ? error_message : "Unable to flush pending changes", rc);
1659 1 : cloudsync_set_sqlstate(data, error_sqlstate);
1660 1 : }
1661 24012 : return rc;
1662 24012 : }
1663 :
1664 : // UPDATE of one column by primary key (pk params 1..npks, value npks+1), prepared on first
1665 : // use. Writing a column of an existing row through the column's upsert proposes a row
1666 : // holding only the primary key and that column; PostgreSQL checks an INSERT policy's
1667 : // WITH CHECK against that proposed row even when the conflict turns it into an update,
1668 : // so a policy on any other column rejects it, as does a NOT NULL constraint. An UPDATE
1669 : // is checked against the real row. Returns NULL when it cannot be prepared.
1670 6368 : static dbvm_t *table_column_update_stmt (cloudsync_context *data, cloudsync_table_context *table, int col_idx) {
1671 6368 : if (!table->col_update_stmt || col_idx < 0 || col_idx >= table->ncols) return NULL;
1672 6368 : if (!table->col_update_stmt[col_idx]) {
1673 188 : const char *colnames[1] = {table->col_name[col_idx]};
1674 188 : char *sql = sql_build_update_pk_and_multi_cols(data, table->name, colnames, 1, table->schema);
1675 188 : if (!sql) return NULL;
1676 188 : databasevm_prepare(data, sql, (void **)&table->col_update_stmt[col_idx], DBFLAG_PERSISTENT);
1677 188 : cloudsync_memory_free(sql);
1678 188 : }
1679 6368 : return table->col_update_stmt[col_idx];
1680 6368 : }
1681 :
1682 3668 : int merge_insert_col (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *col_name, dbvalue_t *col_value, int64_t col_version, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1683 : int index;
1684 3668 : dbvm_t *vm = table_column_lookup(table, col_name, true, &index);
1685 3668 : if (vm == NULL) return cloudsync_set_error(data, "Unable to retrieve column merge precompiled statement in merge_insert_col", DBRES_MISUSE);
1686 :
1687 : // A grow-only set never deletes rows, so tracked metadata means the base row exists:
1688 : // write the column in place instead of through the upsert (see table_column_update_stmt).
1689 3668 : bool update_only = false;
1690 3668 : if (table->algo == table_algo_crdt_gos) {
1691 3150 : int64_t local_cl = merge_get_local_cl(table, pk, pklen);
1692 3150 : dbvm_t *update_vm = (local_cl > 0 && local_cl % 2 == 1) ? table_column_update_stmt(data, table, index) : NULL;
1693 3150 : if (update_vm) { vm = update_vm; update_only = true; }
1694 3150 : }
1695 :
1696 : // INSERT INTO table (pk1, pk2, col_name) VALUES (?, ?, ?) ON CONFLICT DO UPDATE SET col_name=?;"
1697 :
1698 : // bind primary key(s)
1699 3668 : int rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm);
1700 3668 : if (rc < 0) {
1701 0 : cloudsync_set_dberror(data);
1702 0 : dbvm_reset(vm);
1703 0 : return rc;
1704 : }
1705 :
1706 : // bind value (always bind all expected parameters for correct prepared statement handling)
1707 3668 : if (col_value) {
1708 3668 : rc = databasevm_bind_value(vm, table->npks+1, col_value);
1709 3668 : if (rc == DBRES_OK && !update_only) rc = databasevm_bind_value(vm, table->npks+2, col_value);
1710 3668 : } else {
1711 0 : rc = databasevm_bind_null(vm, table->npks+1);
1712 0 : if (rc == DBRES_OK && !update_only) rc = databasevm_bind_null(vm, table->npks+2);
1713 : }
1714 3668 : if (rc != DBRES_OK) {
1715 0 : cloudsync_set_dberror(data);
1716 0 : dbvm_reset(vm);
1717 0 : return rc;
1718 : }
1719 :
1720 : // perform real operation and disable triggers
1721 :
1722 : // in case of GOS we reused the table->col_merge_stmt statement
1723 : // which looks like: INSERT INTO table (pk1, pk2, col_name) VALUES (?, ?, ?) ON CONFLICT DO UPDATE SET col_name=?;"
1724 : // but the UPDATE in the CONFLICT statement would return SQLITE_CONSTRAINT because the trigger raises the error
1725 : // the trick is to disable that trigger before executing the statement
1726 3668 : if (table->algo == table_algo_crdt_gos) table->enabled = 0;
1727 3668 : SYNCBIT_SET(data);
1728 3668 : rc = databasevm_step(vm);
1729 : DEBUG_MERGE("merge_insert(%02x%02x): %s (%d)", data->site_id[UUID_LEN-2], data->site_id[UUID_LEN-1], databasevm_sql(vm), rc);
1730 3668 : bool update_missed = update_only && rc == DBRES_DONE && databasevm_changes(vm) == 0;
1731 3668 : dbvm_reset(vm);
1732 3668 : if (update_missed) {
1733 : // The row is not there to update — gone, or hidden by a policy's USING clause,
1734 : // which an UPDATE skips silently. Fall back to the upsert, which inserts a missing
1735 : // row or reports the policy, instead of recording a write that never happened.
1736 0 : vm = table_column_lookup(table, col_name, true, NULL);
1737 0 : rc = vm ? pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm) : DBRES_MISUSE;
1738 0 : if (rc >= 0 && vm) {
1739 0 : rc = col_value ? databasevm_bind_value(vm, table->npks+1, col_value) : databasevm_bind_null(vm, table->npks+1);
1740 0 : if (rc == DBRES_OK) rc = col_value ? databasevm_bind_value(vm, table->npks+2, col_value) : databasevm_bind_null(vm, table->npks+2);
1741 0 : if (rc == DBRES_OK) rc = databasevm_step(vm);
1742 0 : dbvm_reset(vm);
1743 0 : }
1744 0 : }
1745 3668 : SYNCBIT_RESET(data);
1746 3668 : if (table->algo == table_algo_crdt_gos) table->enabled = 1;
1747 :
1748 3668 : if (rc != DBRES_DONE) {
1749 0 : cloudsync_set_dberror(data);
1750 0 : return rc;
1751 : }
1752 :
1753 3668 : return merge_set_winner_clock(data, table, pk, pklen, col_name, col_version, db_version, site_id, site_len, seq, rowid);
1754 3668 : }
1755 :
1756 207 : int merge_delete (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, const char *colname, int64_t cl, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1757 207 : int rc = DBRES_OK;
1758 :
1759 : // reset return value
1760 207 : *rowid = 0;
1761 :
1762 : // bind pk
1763 207 : dbvm_t *vm = table->real_merge_delete_stmt;
1764 207 : rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm);
1765 207 : if (rc < 0) {
1766 0 : rc = cloudsync_set_dberror(data);
1767 0 : dbvm_reset(vm);
1768 0 : return rc;
1769 : }
1770 :
1771 : // perform real operation and disable triggers
1772 207 : SYNCBIT_SET(data);
1773 207 : rc = databasevm_step(vm);
1774 : DEBUG_MERGE("merge_delete(%02x%02x): %s (%d)", data->site_id[UUID_LEN-2], data->site_id[UUID_LEN-1], databasevm_sql(vm), rc);
1775 207 : dbvm_reset(vm);
1776 207 : SYNCBIT_RESET(data);
1777 207 : if (rc == DBRES_DONE) rc = DBRES_OK;
1778 207 : if (rc != DBRES_OK) {
1779 0 : cloudsync_set_dberror(data);
1780 0 : return rc;
1781 : }
1782 :
1783 207 : rc = merge_set_winner_clock(data, table, pk, pklen, colname, cl, db_version, site_id, site_len, seq, rowid);
1784 207 : if (rc != DBRES_OK) return rc;
1785 :
1786 : // drop clocks _after_ setting the winner clock so we don't lose track of the max db_version!!
1787 : // this must never come before `set_winner_clock`
1788 207 : vm = table->meta_merge_delete_drop;
1789 207 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1790 207 : if (rc == DBRES_OK) rc = databasevm_step(vm);
1791 207 : dbvm_reset(vm);
1792 :
1793 207 : if (rc == DBRES_DONE) rc = DBRES_OK;
1794 207 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1795 207 : return rc;
1796 207 : }
1797 :
1798 265 : int merge_zeroclock_on_resurrect(cloudsync_table_context *table, int64_t db_version, const char *pk, int pklen) {
1799 265 : dbvm_t *vm = table->meta_zero_clock_stmt;
1800 :
1801 265 : int rc = databasevm_bind_int(vm, 1, db_version);
1802 265 : if (rc != DBRES_OK) goto cleanup;
1803 :
1804 265 : rc = databasevm_bind_blob(vm, 2, (const void *)pk, pklen);
1805 265 : if (rc != DBRES_OK) goto cleanup;
1806 :
1807 265 : rc = databasevm_step(vm);
1808 265 : if (rc == DBRES_DONE) rc = DBRES_OK;
1809 :
1810 : cleanup:
1811 265 : if (rc != DBRES_OK) cloudsync_set_dberror(table->context);
1812 265 : dbvm_reset(vm);
1813 265 : return rc;
1814 : }
1815 :
1816 : // executed only if insert_cl == local_cl
1817 53556 : int merge_did_cid_win (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, dbvalue_t *insert_value, const char *site_id, int site_len, const char *col_name, int64_t col_version, bool *didwin_flag) {
1818 :
1819 53556 : if (col_name == NULL) col_name = CLOUDSYNC_TOMBSTONE_VALUE;
1820 :
1821 : int64_t local_version;
1822 53556 : int rc = merge_get_col_version(table, col_name, pk, pklen, &local_version);
1823 53556 : if (rc == DBRES_DONE) {
1824 : // no rows returned, the incoming change wins if there's nothing there locally
1825 24178 : *didwin_flag = true;
1826 24178 : return DBRES_OK;
1827 : }
1828 29378 : if (rc != DBRES_OK) return rc;
1829 :
1830 : // rc == DBRES_OK, means that a row with a version exists
1831 29378 : if (local_version != col_version) {
1832 3595 : if (col_version > local_version) {*didwin_flag = true; return DBRES_OK;}
1833 746 : if (col_version < local_version) {*didwin_flag = false; return DBRES_OK;}
1834 0 : }
1835 :
1836 : // rc == DBRES_ROW and col_version == local_version, need to compare values
1837 :
1838 : // retrieve col_value precompiled statement
1839 25783 : bool is_block_col = block_is_block_colname(col_name) && table_has_block_cols(table);
1840 : dbvm_t *vm;
1841 25783 : if (is_block_col) {
1842 : // Block column: read value from blocks table (pk + col_name bindings)
1843 2463 : vm = table_block_value_read_stmt(table);
1844 2463 : if (!vm) return cloudsync_set_error(data, "Unable to retrieve block value read statement in merge_did_cid_win", DBRES_ERROR);
1845 2463 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1846 2463 : if (rc != DBRES_OK) { dbvm_reset(vm); return cloudsync_set_dberror(data); }
1847 2463 : rc = databasevm_bind_text(vm, 2, col_name, -1);
1848 2463 : if (rc != DBRES_OK) { dbvm_reset(vm); return cloudsync_set_dberror(data); }
1849 2463 : } else {
1850 23320 : vm = table_column_lookup(table, col_name, false, NULL);
1851 23320 : if (!vm) return cloudsync_set_error(data, "Unable to retrieve column value precompiled statement in merge_did_cid_win", DBRES_ERROR);
1852 :
1853 : // bind primary key values
1854 23320 : rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, (void *)vm);
1855 23320 : if (rc < 0) {
1856 0 : rc = cloudsync_set_dberror(data);
1857 0 : dbvm_reset(vm);
1858 0 : return rc;
1859 : }
1860 : }
1861 :
1862 : // execute vm
1863 : dbvalue_t *local_value;
1864 25783 : rc = databasevm_step(vm);
1865 25783 : if (rc == DBRES_DONE) {
1866 : // meta entry exists but the actual value is missing
1867 : // we should allow the value_compare function to make a decision
1868 : // value_compare has been modified to handle the case where lvalue is NULL
1869 462 : local_value = NULL;
1870 462 : rc = DBRES_OK;
1871 25783 : } else if (rc == DBRES_ROW) {
1872 25321 : local_value = database_column_value(vm, 0);
1873 25321 : rc = DBRES_OK;
1874 25321 : } else {
1875 0 : goto cleanup;
1876 : }
1877 :
1878 : // compare values
1879 25783 : int ret = dbutils_value_compare(insert_value, local_value);
1880 : // reset after compare, otherwise local value would be deallocated
1881 25783 : dbvm_reset(vm);
1882 25783 : vm = NULL;
1883 :
1884 25783 : bool compare_site_id = (ret == 0 && data->merge_equal_values == true);
1885 25783 : if (!compare_site_id) {
1886 25783 : *didwin_flag = (ret > 0);
1887 25783 : goto cleanup;
1888 : }
1889 :
1890 : // values are the same and merge_equal_values is true
1891 0 : vm = table->meta_site_id_stmt;
1892 0 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1893 0 : if (rc != DBRES_OK) goto cleanup;
1894 :
1895 0 : rc = databasevm_bind_text(vm, 2, col_name, -1);
1896 0 : if (rc != DBRES_OK) goto cleanup;
1897 :
1898 0 : rc = databasevm_step(vm);
1899 0 : if (rc == DBRES_ROW) {
1900 0 : const void *local_site_id = database_column_blob(vm, 0, NULL);
1901 0 : if (!local_site_id) {
1902 0 : dbvm_reset(vm);
1903 0 : return cloudsync_set_error(data, "NULL site_id in cloudsync table, table is probably corrupted", DBRES_ERROR);
1904 : }
1905 0 : ret = memcmp(site_id, local_site_id, site_len);
1906 0 : *didwin_flag = (ret > 0);
1907 0 : dbvm_reset(vm);
1908 0 : return DBRES_OK;
1909 : }
1910 :
1911 : // handle error condition here
1912 0 : dbvm_reset(vm);
1913 0 : return cloudsync_set_error(data, "Unable to find site_id for previous change, cloudsync table is probably corrupted", DBRES_ERROR);
1914 :
1915 : cleanup:
1916 25783 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1917 25783 : dbvm_reset(vm);
1918 25783 : return rc;
1919 53556 : }
1920 :
1921 265 : int merge_sentinel_only_insert (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, int64_t cl, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1922 :
1923 : // reset return value
1924 265 : *rowid = 0;
1925 :
1926 265 : if (data->pending_batch == NULL) {
1927 : // Immediate mode: execute base table INSERT
1928 1 : dbvm_t *vm = table->real_merge_sentinel_stmt;
1929 1 : int rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm);
1930 1 : if (rc < 0) {
1931 0 : rc = cloudsync_set_dberror(data);
1932 0 : dbvm_reset(vm);
1933 0 : return rc;
1934 : }
1935 :
1936 1 : SYNCBIT_SET(data);
1937 1 : rc = databasevm_step(vm);
1938 1 : dbvm_reset(vm);
1939 1 : SYNCBIT_RESET(data);
1940 1 : if (rc == DBRES_DONE) rc = DBRES_OK;
1941 1 : if (rc != DBRES_OK) {
1942 0 : cloudsync_set_dberror(data);
1943 0 : return rc;
1944 : }
1945 1 : } else {
1946 : // Batch mode: skip base table INSERT, the batch flush will create the row
1947 264 : merge_pending_batch *batch = data->pending_batch;
1948 264 : batch->sentinel_pending = true;
1949 264 : if (batch->table == NULL) {
1950 240 : batch->table = table;
1951 240 : batch->pk = (char *)cloudsync_memory_alloc(pklen);
1952 240 : if (!batch->pk) return cloudsync_set_error(data, "merge_sentinel_only_insert: out of memory for pk", DBRES_NOMEM);
1953 240 : memcpy(batch->pk, pk, pklen);
1954 240 : batch->pk_len = pklen;
1955 240 : }
1956 : }
1957 :
1958 : // Metadata operations always execute regardless of batch mode
1959 265 : int rc = merge_zeroclock_on_resurrect(table, db_version, pk, pklen);
1960 265 : if (rc != DBRES_OK) return rc;
1961 :
1962 265 : return merge_set_winner_clock(data, table, pk, pklen, NULL, cl, db_version, site_id, site_len, seq, rowid);
1963 265 : }
1964 :
1965 : // MARK: - Block-level merge helpers -
1966 :
1967 : // Store a block value in the blocks table
1968 2840 : static int block_store_value (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *block_colname, dbvalue_t *col_value) {
1969 2840 : dbvm_t *vm = table->block_value_write_stmt;
1970 2840 : if (!vm) return cloudsync_set_error(data, "block_store_value: blocks table not initialized", DBRES_MISUSE);
1971 :
1972 2840 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
1973 2840 : if (rc != DBRES_OK) goto cleanup;
1974 2840 : rc = databasevm_bind_text(vm, 2, block_colname, -1);
1975 2840 : if (rc != DBRES_OK) goto cleanup;
1976 2840 : if (col_value) {
1977 2840 : rc = databasevm_bind_value(vm, 3, col_value);
1978 2840 : } else {
1979 0 : rc = databasevm_bind_null(vm, 3);
1980 : }
1981 2840 : if (rc != DBRES_OK) goto cleanup;
1982 :
1983 2840 : rc = databasevm_step(vm);
1984 2840 : if (rc == DBRES_DONE) rc = DBRES_OK;
1985 :
1986 : cleanup:
1987 2840 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1988 2840 : databasevm_reset(vm);
1989 2840 : return rc;
1990 2840 : }
1991 :
1992 : // Delete a block value from the blocks table
1993 1218 : static int block_delete_value (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *block_colname) {
1994 1218 : dbvm_t *vm = table->block_value_delete_stmt;
1995 1218 : if (!vm) return cloudsync_set_error(data, "block_delete_value: blocks table not initialized", DBRES_MISUSE);
1996 :
1997 1218 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
1998 1218 : if (rc != DBRES_OK) goto cleanup;
1999 1218 : rc = databasevm_bind_text(vm, 2, block_colname, -1);
2000 1218 : if (rc != DBRES_OK) goto cleanup;
2001 :
2002 1218 : rc = databasevm_step(vm);
2003 1218 : if (rc == DBRES_DONE) rc = DBRES_OK;
2004 :
2005 : cleanup:
2006 1218 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
2007 1218 : databasevm_reset(vm);
2008 1218 : return rc;
2009 1218 : }
2010 :
2011 : // Materialize all alive blocks for a base column into the base table
2012 : static int block_materialize_column_ex (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *base_col_name, bool row_exists);
2013 :
2014 : // Every failure of block_materialize_column carries a message naming the stage, column
2015 : // and table: a bare code would reach the caller as a blank error. database_error is
2016 : // false for failures that involve no database call (an allocation), whose message must
2017 : // not pick up an unrelated database error left on the connection.
2018 309 : static int block_materialize_fail (cloudsync_context *data, cloudsync_table_context *table, const char *column,
2019 : const char *stage, int rc, bool database_error) {
2020 : char message[512];
2021 309 : if (!database_error) cloudsync_reset_error(data);
2022 309 : if (rc == DBRES_NOMEM) {
2023 308 : snprintf(message, sizeof(message), "Not enough memory to %s the blocks of column \"%s\" of table \"%s\"", stage, column, table->name);
2024 308 : } else {
2025 1 : snprintf(message, sizeof(message), "Unable to %s the blocks of column \"%s\" of table \"%s\"", stage, column, table->name);
2026 : }
2027 309 : return cloudsync_set_error(data, message, (rc > 0) ? rc : DBRES_ERROR);
2028 : }
2029 :
2030 603 : int block_materialize_column (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *base_col_name) {
2031 603 : return block_materialize_column_ex(data, table, pk, pklen, base_col_name, false);
2032 : }
2033 :
2034 : // row_exists selects a plain UPDATE of the column instead of the column's upsert (see
2035 : // table_column_update_stmt).
2036 4157 : static int block_materialize_column_ex (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *base_col_name, bool row_exists) {
2037 4157 : if (!table->block_list_stmt) return block_materialize_fail(data, table, base_col_name, "read", DBRES_MISUSE, false);
2038 :
2039 : // Find column index and delimiter
2040 4157 : int col_idx = -1;
2041 5880 : for (int i = 0; i < table->ncols; i++) {
2042 5880 : if (strcasecmp(table->col_name[i], base_col_name) == 0) {
2043 4157 : col_idx = i;
2044 4157 : break;
2045 : }
2046 1723 : }
2047 4157 : if (col_idx < 0) return block_materialize_fail(data, table, base_col_name, "locate", DBRES_ERROR, false);
2048 4157 : const char *delimiter = table->col_delimiter[col_idx] ? table->col_delimiter[col_idx] : BLOCK_DEFAULT_DELIMITER;
2049 :
2050 : // Build the LIKE pattern for block col_names: "base_col\x1F%"
2051 4157 : char *like_pattern = block_build_colname(base_col_name, "%");
2052 4157 : if (!like_pattern) return block_materialize_fail(data, table, base_col_name, "read", DBRES_NOMEM, false);
2053 :
2054 : // Query alive blocks from blocks table joined with metadata
2055 : // block_list_stmt: SELECT b.col_value FROM blocks b JOIN meta m
2056 : // ON b.pk = m.pk AND b.col_name = m.col_name
2057 : // WHERE b.pk = ? AND b.col_name LIKE ? AND m.col_version % 2 = 1
2058 : // ORDER BY b.col_name
2059 4156 : dbvm_t *vm = table->block_list_stmt;
2060 4156 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
2061 4156 : if (rc == DBRES_OK) rc = databasevm_bind_text(vm, 2, like_pattern, -1);
2062 : // Bind pk again for the join condition (parameter 3)
2063 4156 : if (rc == DBRES_OK) rc = databasevm_bind_blob(vm, 3, pk, pklen);
2064 4156 : if (rc == DBRES_OK) rc = databasevm_bind_text(vm, 4, like_pattern, -1);
2065 4156 : if (rc != DBRES_OK) {
2066 0 : cloudsync_memory_free(like_pattern);
2067 0 : databasevm_reset(vm);
2068 0 : return block_materialize_fail(data, table, base_col_name, "read", rc, true);
2069 : }
2070 :
2071 : // Collect block values
2072 4156 : const char **block_values = NULL;
2073 4156 : int block_count = 0;
2074 4156 : int block_cap = 0;
2075 :
2076 36351 : while ((rc = databasevm_step(vm)) == DBRES_ROW) {
2077 32198 : const char *value = database_column_text(vm, 0);
2078 32198 : if (block_count >= block_cap) {
2079 4646 : int new_cap = block_cap ? block_cap * 2 : 16;
2080 4646 : const char **new_arr = (const char **)cloudsync_memory_realloc((void *)block_values, (uint64_t)(new_cap * sizeof(char *)));
2081 4646 : if (!new_arr) { rc = DBRES_NOMEM; break; }
2082 4645 : block_values = new_arr;
2083 4645 : block_cap = new_cap;
2084 4645 : }
2085 32197 : block_values[block_count] = value ? cloudsync_string_dup(value) : cloudsync_string_dup("");
2086 32197 : if (!block_values[block_count]) { rc = DBRES_NOMEM; break; }
2087 32195 : block_count++;
2088 : }
2089 4156 : databasevm_reset(vm);
2090 4156 : cloudsync_memory_free(like_pattern);
2091 :
2092 4156 : if (rc != DBRES_DONE && rc != DBRES_OK && rc != DBRES_ROW) {
2093 : // Free collected values
2094 128 : for (int i = 0; i < block_count; i++) cloudsync_memory_free((void *)block_values[i]);
2095 127 : if (block_values) cloudsync_memory_free((void *)block_values);
2096 : // an allocation failing while collecting is not a database read failure
2097 127 : if (rc == DBRES_NOMEM) return block_materialize_fail(data, table, base_col_name, "collect", rc, false);
2098 0 : return block_materialize_fail(data, table, base_col_name, "read", rc, true);
2099 : }
2100 :
2101 : // Materialize text (NULL when no alive blocks)
2102 4029 : char *text = (block_count > 0) ? block_materialize_text(block_values, block_count, delimiter) : NULL;
2103 36223 : for (int i = 0; i < block_count; i++) cloudsync_memory_free((void *)block_values[i]);
2104 4029 : if (block_values) cloudsync_memory_free((void *)block_values);
2105 4029 : if (block_count > 0 && !text) return block_materialize_fail(data, table, base_col_name, "join", DBRES_NOMEM, false);
2106 :
2107 : // Update the base table column via the col_merge_stmt (with triggers disabled)
2108 4028 : dbvm_t *merge_vm = table->col_merge_stmt[col_idx];
2109 4028 : bool update_only = false;
2110 4028 : dbvm_t *update_vm = row_exists ? table_column_update_stmt(data, table, col_idx) : NULL;
2111 4028 : if (update_vm) { merge_vm = update_vm; update_only = true; }
2112 4028 : if (!merge_vm) { cloudsync_memory_free(text); return block_materialize_fail(data, table, base_col_name, "write", DBRES_MISUSE, false); }
2113 :
2114 : // Bind PKs
2115 4028 : rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, merge_vm);
2116 4028 : if (rc < 0) {
2117 0 : cloudsync_memory_free(text);
2118 0 : databasevm_reset(merge_vm);
2119 0 : return block_materialize_fail(data, table, base_col_name, "write", DBRES_ERROR, true);
2120 : }
2121 :
2122 : // Bind the text value twice (INSERT value + ON CONFLICT UPDATE value); the plain
2123 : // UPDATE takes it once
2124 4028 : int npks = table->npks;
2125 4028 : rc = text ? databasevm_bind_text(merge_vm, npks + 1, text, -1) : databasevm_bind_null(merge_vm, npks + 1);
2126 4028 : if (rc == DBRES_OK && !update_only) rc = text ? databasevm_bind_text(merge_vm, npks + 2, text, -1) : databasevm_bind_null(merge_vm, npks + 2);
2127 4028 : if (rc != DBRES_OK) {
2128 0 : cloudsync_memory_free(text);
2129 0 : databasevm_reset(merge_vm);
2130 0 : return block_materialize_fail(data, table, base_col_name, "write", rc, true);
2131 : }
2132 :
2133 : // Execute with triggers disabled
2134 4028 : table->enabled = 0;
2135 4028 : SYNCBIT_SET(data);
2136 4028 : rc = databasevm_step(merge_vm);
2137 4028 : bool update_missed = update_only && rc == DBRES_DONE && databasevm_changes(merge_vm) == 0;
2138 4028 : databasevm_reset(merge_vm);
2139 4028 : if (update_missed) {
2140 : // Nothing to update — the row is gone, or a policy's USING clause hides it and the
2141 : // UPDATE skipped it silently. Fall back to the upsert, which inserts a missing row
2142 : // or reports the policy, rather than leave the column unwritten without a word.
2143 2 : merge_vm = table->col_merge_stmt[col_idx];
2144 2 : rc = merge_vm ? pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, merge_vm) : DBRES_MISUSE;
2145 2 : if (rc >= 0 && merge_vm) {
2146 2 : rc = text ? databasevm_bind_text(merge_vm, npks + 1, text, -1) : databasevm_bind_null(merge_vm, npks + 1);
2147 2 : if (rc == DBRES_OK) rc = text ? databasevm_bind_text(merge_vm, npks + 2, text, -1) : databasevm_bind_null(merge_vm, npks + 2);
2148 2 : if (rc == DBRES_OK) rc = databasevm_step(merge_vm);
2149 2 : databasevm_reset(merge_vm);
2150 2 : }
2151 2 : }
2152 4028 : SYNCBIT_RESET(data);
2153 4028 : table->enabled = 1;
2154 :
2155 4028 : cloudsync_memory_free(text);
2156 :
2157 4028 : if (rc == DBRES_DONE) rc = DBRES_OK;
2158 4028 : if (rc != DBRES_OK) return block_materialize_fail(data, table, base_col_name, "write", (rc > 0) ? rc : DBRES_ERROR, true);
2159 3848 : return DBRES_OK;
2160 4157 : }
2161 :
2162 : // Accessor for has_block_cols flag
2163 14698 : bool table_has_block_cols (cloudsync_table_context *table) {
2164 14698 : return table && table->has_block_cols;
2165 : }
2166 :
2167 : // Get block column algo for a given column index
2168 15458 : col_algo_t table_col_algo (cloudsync_table_context *table, int index) {
2169 15458 : if (!table || !table->col_algo || index < 0 || index >= table->ncols) return col_algo_normal;
2170 15458 : return table->col_algo[index];
2171 15458 : }
2172 :
2173 : // Get block delimiter for a given column index
2174 959 : const char *table_col_delimiter (cloudsync_table_context *table, int index) {
2175 959 : if (!table || !table->col_delimiter || index < 0 || index >= table->ncols) return BLOCK_DEFAULT_DELIMITER;
2176 959 : return table->col_delimiter[index] ? table->col_delimiter[index] : BLOCK_DEFAULT_DELIMITER;
2177 959 : }
2178 :
2179 : // Block column struct accessors (for use outside cloudsync.c where struct is opaque)
2180 14698 : dbvm_t *table_block_value_read_stmt (cloudsync_table_context *table) { return table ? table->block_value_read_stmt : NULL; }
2181 2485 : dbvm_t *table_block_value_write_stmt (cloudsync_table_context *table) { return table ? table->block_value_write_stmt : NULL; }
2182 960 : const char *table_blocks_ref (cloudsync_table_context *table) { return table ? table->blocks_ref : NULL; }
2183 :
2184 3 : void table_set_col_delimiter (cloudsync_table_context *table, int col_idx, const char *delimiter) {
2185 3 : if (!table || !table->col_delimiter || col_idx < 0 || col_idx >= table->ncols) return;
2186 3 : if (table->col_delimiter[col_idx]) cloudsync_memory_free(table->col_delimiter[col_idx]);
2187 3 : table->col_delimiter[col_idx] = delimiter ? cloudsync_string_dup(delimiter) : NULL;
2188 3 : }
2189 :
2190 : // Find column index by name
2191 940 : int table_col_index (cloudsync_table_context *table, const char *col_name) {
2192 940 : if (!table || !col_name) return -1;
2193 1190 : for (int i = 0; i < table->ncols; i++) {
2194 1190 : if (strcasecmp(table->col_name[i], col_name) == 0) return i;
2195 250 : }
2196 0 : return -1;
2197 940 : }
2198 :
2199 54912 : int merge_insert (cloudsync_context *data, cloudsync_table_context *table, const char *insert_pk, int insert_pk_len, int64_t insert_cl, const char *insert_name, dbvalue_t *insert_value, int64_t insert_col_version, int64_t insert_db_version, const char *insert_site_id, int insert_site_id_len, int64_t insert_seq, int64_t *rowid) {
2200 : // Handle DWS and AWS algorithms here
2201 : // Delete-Wins Set (DWS): table_algo_crdt_dws
2202 : // Add-Wins Set (AWS): table_algo_crdt_aws
2203 :
2204 : // Causal-Length Set (CLS) Algorithm (default)
2205 :
2206 : // compute the local causal length for the row based on the primary key
2207 : // the causal length is used to determine the order of operations and resolve conflicts.
2208 54912 : int64_t local_cl = merge_get_local_cl(table, insert_pk, insert_pk_len);
2209 54912 : if (local_cl < 0) return cloudsync_set_error(data, "Unable to compute local causal length", DBRES_ERROR);
2210 :
2211 : // if the incoming causal length is older than the local causal length, we can safely ignore it
2212 : // because the local changes are more recent
2213 54912 : if (insert_cl < local_cl) return DBRES_OK;
2214 :
2215 : // check if the operation is a delete by examining the causal length
2216 : // even causal lengths typically signify delete operations
2217 54694 : bool is_delete = (insert_cl % 2 == 0);
2218 54694 : if (is_delete) {
2219 : // if it's a delete, check if the local state is at the same causal length
2220 : // if it is, no further action is needed
2221 648 : if (local_cl == insert_cl) return DBRES_OK;
2222 :
2223 : // perform a delete merge if the causal length is newer than the local one
2224 414 : int rc = merge_delete(data, table, insert_pk, insert_pk_len, insert_name, insert_col_version,
2225 207 : insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2226 207 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_delete", rc);
2227 207 : return rc;
2228 : }
2229 :
2230 : // if the operation is a sentinel-only insert (indicating a new row or resurrected row with no column update), handle it separately.
2231 54046 : bool is_sentinel_only = (strcmp(insert_name, CLOUDSYNC_TOMBSTONE_VALUE) == 0);
2232 54046 : if (is_sentinel_only) {
2233 490 : if (local_cl == insert_cl) return DBRES_OK;
2234 :
2235 : // perform a sentinel-only insert to track the existence of the row
2236 528 : int rc = merge_sentinel_only_insert(data, table, insert_pk, insert_pk_len, insert_col_version,
2237 264 : insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2238 264 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_sentinel_only_insert", rc);
2239 264 : return rc;
2240 : }
2241 :
2242 : // from this point I can be sure that insert_name is not sentinel
2243 :
2244 : // handle the case where a row is being resurrected (e.g., after a delete, a new insert for the same row)
2245 : // odd causal lengths can "resurrect" rows
2246 53556 : bool needs_resurrect = (insert_cl > local_cl && insert_cl % 2 == 1);
2247 53556 : bool row_exists_locally = local_cl != 0;
2248 :
2249 : // if a resurrection is needed, insert a sentinel to mark the row as alive
2250 : // this handles out-of-order deliveries where the row was deleted and is now being re-inserted
2251 53556 : if (needs_resurrect && (row_exists_locally || (!row_exists_locally && insert_cl > 1))) {
2252 2 : int rc = merge_sentinel_only_insert(data, table, insert_pk, insert_pk_len, insert_cl,
2253 1 : insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2254 1 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to perform merge_sentinel_only_insert", rc);
2255 1 : }
2256 :
2257 : // at this point, we determine whether the incoming change wins based on causal length
2258 : // this can be due to a resurrection, a non-existent local row, or a conflict resolution
2259 53556 : bool flag = false;
2260 53556 : int rc = merge_did_cid_win(data, table, insert_pk, insert_pk_len, insert_value, insert_site_id, insert_site_id_len, insert_name, insert_col_version, &flag);
2261 53556 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to perform merge_did_cid_win", rc);
2262 :
2263 : // check if the incoming change wins and should be applied
2264 53556 : bool does_cid_win = ((needs_resurrect) || (!row_exists_locally) || (flag));
2265 53556 : if (!does_cid_win) return DBRES_OK;
2266 :
2267 : // Block-level LWW: if the incoming col_name is a block entry (contains \x1F),
2268 : // bypass the normal base-table write and instead store the value in the blocks table.
2269 : // The base table column will be materialized from all alive blocks.
2270 27895 : if (block_is_block_colname(insert_name) && table->has_block_cols) {
2271 : // Store or delete block value in blocks table depending on tombstone status
2272 3674 : if (insert_col_version % 2 == 0) {
2273 : // Tombstone: remove from blocks table
2274 834 : rc = block_delete_value(data, table, insert_pk, insert_pk_len, insert_name);
2275 834 : } else {
2276 2840 : rc = block_store_value(data, table, insert_pk, insert_pk_len, insert_name, insert_value);
2277 : }
2278 3674 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to store/delete block value", rc);
2279 :
2280 : // Set winner clock in metadata
2281 7108 : rc = merge_set_winner_clock(data, table, insert_pk, insert_pk_len, insert_name,
2282 3554 : insert_col_version, insert_db_version,
2283 3554 : insert_site_id, insert_site_id_len, insert_seq, rowid);
2284 3554 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to set winner clock for block", rc);
2285 :
2286 : // Materialize the full column from blocks into the base table
2287 3554 : char *base_col = block_extract_base_colname(insert_name);
2288 3554 : if (base_col) {
2289 3554 : rc = block_materialize_column_ex(data, table, insert_pk, insert_pk_len, base_col, (local_cl % 2 == 1) && !needs_resurrect);
2290 3554 : cloudsync_memory_free(base_col);
2291 : // the failure already names the stage, column and table
2292 3554 : if (rc != DBRES_OK) return rc;
2293 3554 : }
2294 :
2295 3554 : return DBRES_OK;
2296 : }
2297 :
2298 : // perform the final column insert or update if the incoming change wins
2299 24221 : if (data->pending_batch) {
2300 : // Propagate row_exists_locally to the batch on the first winning column.
2301 : // This lets merge_flush_pending choose UPDATE vs INSERT ON CONFLICT,
2302 : // which matters when RLS policies reference columns not in the payload.
2303 23703 : if (data->pending_batch->table == NULL) {
2304 9432 : data->pending_batch->row_exists = row_exists_locally;
2305 9432 : }
2306 23703 : rc = merge_pending_add(data, table, insert_pk, insert_pk_len, insert_name, insert_value, insert_col_version, insert_db_version, insert_site_id, insert_site_id_len, insert_seq);
2307 23703 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_pending_add", rc);
2308 23703 : } else {
2309 518 : rc = merge_insert_col(data, table, insert_pk, insert_pk_len, insert_name, insert_value, insert_col_version, insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2310 518 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_insert_col", rc);
2311 : }
2312 :
2313 24221 : return rc;
2314 54912 : }
2315 :
2316 : // MARK: - Block column setup -
2317 :
2318 : // Migrate existing tracked rows to block format when block-level LWW is first enabled on a column.
2319 : // Scans the metadata table for alive rows with the plain col_name entry (not yet block entries),
2320 : // reads each row's current value from the base table, splits it into blocks, and inserts
2321 : // the block entries into both the blocks table and the metadata table.
2322 : // Uses INSERT OR IGNORE semantics so the operation is safe to call multiple times.
2323 331 : static int block_migrate_existing_rows (cloudsync_context *data, cloudsync_table_context *table, int col_idx) {
2324 331 : const char *col_name = table->col_name[col_idx];
2325 331 : if (!col_name || !table->meta_ref || !table->blocks_ref) return DBRES_OK;
2326 :
2327 331 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
2328 :
2329 : // Phase 1: collect all existing PKs that have an alive regular col_name entry
2330 : // AND do not yet have any entries in the blocks table for this column.
2331 : // The NOT IN filter makes this idempotent: rows that were already migrated
2332 : // (or had their blocks created via INSERT) are skipped on subsequent calls.
2333 : // We collect PKs before writing so that writes to the metadata table (Phase 2)
2334 : // do not perturb the read cursor on the same table.
2335 331 : char *like_pattern = block_build_colname(col_name, "%");
2336 331 : if (!like_pattern) return DBRES_NOMEM;
2337 :
2338 331 : char *scan_sql = cloudsync_memory_mprintf(SQL_META_SCAN_COL_FOR_MIGRATION, table->meta_ref, table->blocks_ref);
2339 331 : if (!scan_sql) { cloudsync_memory_free(like_pattern); return DBRES_NOMEM; }
2340 331 : dbvm_t *scan_vm = NULL;
2341 331 : int rc = databasevm_prepare(data, scan_sql, &scan_vm, 0);
2342 331 : cloudsync_memory_free(scan_sql);
2343 331 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); return rc; }
2344 :
2345 331 : rc = databasevm_bind_text(scan_vm, 1, col_name, -1);
2346 331 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_finalize(scan_vm); return rc; }
2347 : // Bind like_pattern as ?2 and keep it alive until after all scan steps complete,
2348 : // because databasevm_bind_text uses SQLITE_STATIC (no copy).
2349 331 : rc = databasevm_bind_text(scan_vm, 2, like_pattern, -1);
2350 331 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_finalize(scan_vm); return rc; }
2351 :
2352 : // Collect pk blobs into a dynamically-grown array of owned copies
2353 331 : void **pks = NULL;
2354 331 : size_t *pklens = NULL;
2355 331 : int pk_count = 0;
2356 331 : int pk_cap = 0;
2357 :
2358 335 : while ((rc = databasevm_step(scan_vm)) == DBRES_ROW) {
2359 4 : size_t pklen = 0;
2360 4 : const void *pk = database_column_blob(scan_vm, 0, &pklen);
2361 4 : if (!pk || pklen == 0) continue;
2362 :
2363 4 : if (pk_count >= pk_cap) {
2364 2 : int new_cap = pk_cap ? pk_cap * 2 : 8;
2365 2 : void **new_pks = (void **)cloudsync_memory_realloc(pks, (uint64_t)(new_cap * sizeof(void *)));
2366 2 : size_t *new_pklens = (size_t *)cloudsync_memory_realloc(pklens, (uint64_t)(new_cap * sizeof(size_t)));
2367 2 : if (!new_pks || !new_pklens) {
2368 0 : for (int i = 0; i < pk_count; i++) cloudsync_memory_free((new_pks ? new_pks : pks)[i]);
2369 0 : cloudsync_memory_free(new_pks ? new_pks : pks);
2370 0 : cloudsync_memory_free(new_pklens ? new_pklens : pklens);
2371 0 : databasevm_finalize(scan_vm);
2372 0 : cloudsync_memory_free(like_pattern);
2373 0 : return DBRES_NOMEM;
2374 : }
2375 2 : pks = new_pks;
2376 2 : pklens = new_pklens;
2377 2 : pk_cap = new_cap;
2378 2 : }
2379 :
2380 4 : pks[pk_count] = cloudsync_memory_alloc((uint64_t)pklen);
2381 4 : if (!pks[pk_count]) { rc = DBRES_NOMEM; break; }
2382 4 : memcpy(pks[pk_count], pk, pklen);
2383 4 : pklens[pk_count] = pklen;
2384 4 : pk_count++;
2385 : }
2386 :
2387 331 : databasevm_finalize(scan_vm);
2388 331 : cloudsync_memory_free(like_pattern); // safe to free after scan_vm is finalized
2389 331 : if (rc != DBRES_DONE && rc != DBRES_OK) {
2390 0 : for (int i = 0; i < pk_count; i++) cloudsync_memory_free(pks[i]);
2391 0 : cloudsync_memory_free(pks);
2392 0 : cloudsync_memory_free(pklens);
2393 0 : return rc;
2394 : }
2395 :
2396 331 : if (pk_count == 0) {
2397 329 : cloudsync_memory_free(pks);
2398 329 : cloudsync_memory_free(pklens);
2399 329 : return DBRES_OK;
2400 : }
2401 :
2402 : // Reuse the checked block writer; the scan above excludes migrated rows.
2403 : // As in local_block_insert, every failure carries a message: the migration aborts
2404 : // on one, and a bare code would surface as a blank error. Aborting is recoverable
2405 : // — the Phase 1 scan skips already-migrated rows, so a re-run resumes. A row that
2406 : // cannot be read back is skipped rather than failed (see below).
2407 : char errmsg[512];
2408 2 : dbvm_t *val_vm = table_column_lookup(table, col_name, false, NULL);
2409 2 : if (!val_vm) {
2410 0 : snprintf(errmsg, sizeof(errmsg), "Missing value statement for block column \"%s\" of table \"%s\"", col_name, table->name);
2411 0 : rc = cloudsync_set_error(data, errmsg, DBRES_MISUSE);
2412 0 : } else {
2413 2 : rc = DBRES_OK;
2414 : }
2415 6 : for (int p = 0; p < pk_count && rc == DBRES_OK; p++) {
2416 4 : rc = pk_decode_prikey(pks[p], pklens[p], pk_decode_bind_callback, val_vm);
2417 4 : if (rc < 0) {
2418 0 : snprintf(errmsg, sizeof(errmsg), "Unable to decode the primary key of a row in \"%s\" while migrating block column \"%s\"", table->name, col_name);
2419 0 : rc = cloudsync_set_error(data, errmsg, DBRES_ERROR);
2420 0 : databasevm_reset(val_vm);
2421 0 : break;
2422 : }
2423 4 : rc = databasevm_step(val_vm);
2424 4 : if (rc == DBRES_ROW) {
2425 3 : const char *text = database_column_text(val_vm, 0);
2426 3 : bool has_text = text != NULL;
2427 3 : char *copy = text ? cloudsync_string_dup(text) : NULL;
2428 3 : databasevm_reset(val_vm);
2429 3 : if (has_text && !copy) {
2430 0 : snprintf(errmsg, sizeof(errmsg), "Not enough memory to migrate block column \"%s\" of table \"%s\"", col_name, table->name);
2431 0 : rc = cloudsync_set_error(data, errmsg, DBRES_NOMEM);
2432 0 : } else {
2433 3 : rc = DBRES_OK;
2434 : }
2435 3 : if (rc == DBRES_OK && copy)
2436 3 : rc = local_block_update(data, table, pks[p], pklens[p], col_idx, copy, db_version, true);
2437 3 : cloudsync_memory_free(copy);
2438 4 : } else if (rc == DBRES_DONE) {
2439 : // The scan reads the metadata table, not the base table, so it can return a
2440 : // pk whose row is gone: deleted while sync was disabled, say, or hidden from
2441 : // this session by a row-level security SELECT policy. There is no current
2442 : // value to split, and failing here would make the column impossible to
2443 : // convert. Skip it as the migration always has; if the row reappears, its
2444 : // next local write creates the blocks.
2445 1 : rc = DBRES_OK;
2446 1 : }
2447 4 : databasevm_reset(val_vm);
2448 4 : }
2449 6 : for (int i = 0; i < pk_count; i++) cloudsync_memory_free(pks[i]);
2450 2 : cloudsync_memory_free(pks);
2451 2 : cloudsync_memory_free(pklens);
2452 2 : return rc;
2453 331 : }
2454 :
2455 334 : int cloudsync_setup_block_column (cloudsync_context *data, const char *table_name, const char *col_name, const char *delimiter, bool persist) {
2456 334 : cloudsync_table_context *table = table_lookup(data, table_name);
2457 334 : if (!table) return cloudsync_set_error(data, "cloudsync_setup_block_column: table not found", DBRES_ERROR);
2458 :
2459 : // Find column index
2460 334 : int col_idx = table_col_index(table, col_name);
2461 334 : if (col_idx < 0) {
2462 : char buf[1024];
2463 0 : snprintf(buf, sizeof(buf), "cloudsync_setup_block_column: column '%s' not found in table '%s'", col_name, table_name);
2464 0 : return cloudsync_set_error(data, buf, DBRES_ERROR);
2465 : }
2466 :
2467 : // Set column algo
2468 334 : table->col_algo[col_idx] = col_algo_block;
2469 334 : table->has_block_cols = true;
2470 :
2471 : // Set delimiter (can be NULL for default)
2472 334 : if (table->col_delimiter[col_idx]) {
2473 0 : cloudsync_memory_free(table->col_delimiter[col_idx]);
2474 0 : table->col_delimiter[col_idx] = NULL;
2475 0 : }
2476 334 : if (delimiter) {
2477 1 : table->col_delimiter[col_idx] = cloudsync_string_dup(delimiter);
2478 1 : }
2479 :
2480 : // Create blocks table if not already done
2481 334 : if (!table->blocks_ref) {
2482 327 : table->blocks_ref = database_build_blocks_ref(table->schema, table->name);
2483 327 : if (!table->blocks_ref) return DBRES_NOMEM;
2484 :
2485 : // CREATE TABLE IF NOT EXISTS
2486 327 : char *sql = cloudsync_memory_mprintf(SQL_BLOCKS_CREATE_TABLE, table->blocks_ref);
2487 327 : if (!sql) return DBRES_NOMEM;
2488 :
2489 327 : int rc = database_exec(data, sql);
2490 327 : cloudsync_memory_free(sql);
2491 327 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to create blocks table", rc);
2492 :
2493 : // Prepare block statements
2494 : // Write: upsert into blocks (pk, col_name, col_value)
2495 327 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_UPSERT, table->blocks_ref);
2496 327 : if (!sql) return DBRES_NOMEM;
2497 327 : rc = databasevm_prepare(data, sql, (void **)&table->block_value_write_stmt, DBFLAG_PERSISTENT);
2498 327 : cloudsync_memory_free(sql);
2499 327 : if (rc != DBRES_OK) return rc;
2500 :
2501 : // Read: SELECT col_value FROM blocks WHERE pk = ? AND col_name = ?
2502 327 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_SELECT, table->blocks_ref);
2503 327 : if (!sql) return DBRES_NOMEM;
2504 327 : rc = databasevm_prepare(data, sql, (void **)&table->block_value_read_stmt, DBFLAG_PERSISTENT);
2505 327 : cloudsync_memory_free(sql);
2506 327 : if (rc != DBRES_OK) return rc;
2507 :
2508 : // Delete: DELETE FROM blocks WHERE pk = ? AND col_name = ?
2509 327 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_DELETE, table->blocks_ref);
2510 327 : if (!sql) return DBRES_NOMEM;
2511 327 : rc = databasevm_prepare(data, sql, (void **)&table->block_value_delete_stmt, DBFLAG_PERSISTENT);
2512 327 : cloudsync_memory_free(sql);
2513 327 : if (rc != DBRES_OK) return rc;
2514 :
2515 : // List alive blocks for materialization
2516 327 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_LIST_ALIVE, table->blocks_ref, table->meta_ref);
2517 327 : if (!sql) return DBRES_NOMEM;
2518 327 : rc = databasevm_prepare(data, sql, (void **)&table->block_list_stmt, DBFLAG_PERSISTENT);
2519 327 : cloudsync_memory_free(sql);
2520 327 : if (rc != DBRES_OK) return rc;
2521 327 : }
2522 :
2523 : // Persist settings (skipped when called from the settings loader, since
2524 : // writing to cloudsync_table_settings while sqlite3_exec is iterating it
2525 : // re-feeds the rewritten row to the cursor and causes an infinite loop).
2526 334 : if (persist) {
2527 331 : int rc = dbutils_table_settings_set_key_value(data, table_name, col_name, "algo", "block");
2528 331 : if (rc != DBRES_OK) return rc;
2529 :
2530 331 : if (delimiter) {
2531 0 : rc = dbutils_table_settings_set_key_value(data, table_name, col_name, "delimiter", delimiter);
2532 0 : if (rc != DBRES_OK) return rc;
2533 0 : }
2534 :
2535 : // Migrate any existing tracked rows: populate the blocks table and metadata with
2536 : // block entries derived from the current column value, so that subsequent UPDATE
2537 : // operations can diff against the real existing state instead of treating everything
2538 : // as new, and so this node participates correctly in LWW conflict resolution.
2539 331 : rc = block_migrate_existing_rows(data, table, col_idx);
2540 331 : if (rc != DBRES_OK) return rc;
2541 331 : }
2542 :
2543 334 : return DBRES_OK;
2544 334 : }
2545 :
2546 : // MARK: - Private -
2547 :
2548 772 : bool cloudsync_config_exists (cloudsync_context *data) {
2549 772 : return database_internal_table_exists(data, CLOUDSYNC_SITEID_NAME) == true;
2550 : }
2551 :
2552 2138 : bool cloudsync_context_is_initialized (cloudsync_context *data) {
2553 : // A fully initialized context has its persistent "is the DB stale" probe
2554 : // prepared. cloudsync_context_init prepares data_version_stmt (via
2555 : // cloudsync_add_dbvms) only after the cloudsync_site_id table exists, so
2556 : // a non-NULL pointer means cloudsync_init has been called at least once
2557 : // on this connection. Used to produce actionable error messages when
2558 : // callers hit a function before calling cloudsync_init.
2559 2138 : return data != NULL && data->data_version_stmt != NULL;
2560 : }
2561 :
2562 790 : cloudsync_context *cloudsync_context_create (void *db) {
2563 790 : cloudsync_context *data = (cloudsync_context *)cloudsync_memory_zeroalloc((uint64_t)(sizeof(cloudsync_context)));
2564 790 : if (!data) return NULL;
2565 : DEBUG_SETTINGS("cloudsync_context_create %p", data);
2566 :
2567 790 : data->libversion = CLOUDSYNC_VERSION;
2568 790 : data->pending_db_version = CLOUDSYNC_VALUE_NOTSET;
2569 : #if CLOUDSYNC_DEBUG
2570 : data->debug = 1;
2571 : #endif
2572 :
2573 : // allocate space for 64 tables (it can grow if needed)
2574 790 : uint64_t mem_needed = (uint64_t)(CLOUDSYNC_INIT_NTABLES * sizeof(cloudsync_table_context *));
2575 790 : data->tables = (cloudsync_table_context **)cloudsync_memory_zeroalloc(mem_needed);
2576 790 : if (!data->tables) {cloudsync_memory_free(data); return NULL;}
2577 :
2578 790 : data->tables_cap = CLOUDSYNC_INIT_NTABLES;
2579 790 : data->tables_count = 0;
2580 790 : data->db = db;
2581 :
2582 : // SQLite exposes col_value as ANY, but other databases require a concrete type.
2583 : // In PostgreSQL we expose col_value as bytea, which holds the pk-encoded value bytes (type + data).
2584 : // Because col_value is already encoded, we skip decoding this field and pass it through as bytea.
2585 : // It is decoded to the target column type just before applying changes to the base table.
2586 790 : data->skip_decode_idx = (db == NULL) ? CLOUDSYNC_PK_INDEX_COLVALUE : -1;
2587 :
2588 790 : return data;
2589 790 : }
2590 :
2591 790 : void cloudsync_context_free (void *ctx) {
2592 790 : cloudsync_context *data = (cloudsync_context *)ctx;
2593 : DEBUG_SETTINGS("cloudsync_context_free %p", data);
2594 790 : if (!data) return;
2595 :
2596 : // free all table contexts and prepared statements
2597 790 : cloudsync_terminate(data);
2598 :
2599 790 : cloudsync_memory_free(data->tables);
2600 790 : cloudsync_memory_free(data->stream_values);
2601 790 : cloudsync_memory_free(data);
2602 790 : }
2603 :
2604 903 : const char *cloudsync_context_init (cloudsync_context *data) {
2605 903 : if (!data) return NULL;
2606 :
2607 : // perform init just the first time, if the site_id field is not set.
2608 : // The data->site_id value could exists while settings tables don't exists if the
2609 : // cloudsync_context_init was previously called in init transaction that was rolled back
2610 : // because of an error during the init process.
2611 903 : if (data->site_id[0] == 0 || !database_internal_table_exists(data, CLOUDSYNC_SITEID_NAME)) {
2612 769 : if (dbutils_settings_init(data) != DBRES_OK) return NULL;
2613 769 : if (cloudsync_add_dbvms(data) != DBRES_OK) return NULL;
2614 769 : if (cloudsync_load_siteid(data) != DBRES_OK) return NULL;
2615 769 : data->schema_hash = database_schema_hash(data);
2616 769 : }
2617 :
2618 903 : return (const char *)data->site_id;
2619 903 : }
2620 :
2621 3528 : void cloudsync_sync_key (cloudsync_context *data, const char *key, const char *value) {
2622 : DEBUG_SETTINGS("cloudsync_sync_key key: %s value: %s", key, value);
2623 :
2624 : // sync data
2625 3528 : if (strcmp(key, CLOUDSYNC_KEY_SCHEMAVERSION) == 0) {
2626 769 : data->schema_version = (int)strtol(value, NULL, 0);
2627 769 : return;
2628 : }
2629 :
2630 2759 : if (strcmp(key, CLOUDSYNC_KEY_DEBUG) == 0) {
2631 0 : data->debug = 0;
2632 0 : if (value && (value[0] != 0) && (value[0] != '0')) data->debug = 1;
2633 0 : return;
2634 : }
2635 :
2636 2759 : if (strcmp(key, CLOUDSYNC_KEY_SCHEMA) == 0) {
2637 0 : cloudsync_set_schema(data, value);
2638 0 : return;
2639 : }
2640 3528 : }
2641 :
2642 : #if 0
2643 : void cloudsync_sync_table_key(cloudsync_context *data, const char *table, const char *column, const char *key, const char *value) {
2644 : DEBUG_SETTINGS("cloudsync_sync_table_key table: %s column: %s key: %s value: %s", table, column, key, value);
2645 : // Unused in this version
2646 : return;
2647 : }
2648 : #endif
2649 :
2650 16293 : int cloudsync_commit_hook (void *ctx) {
2651 16293 : cloudsync_context *data = (cloudsync_context *)ctx;
2652 :
2653 16293 : data->db_version = data->pending_db_version;
2654 16293 : data->pending_db_version = CLOUDSYNC_VALUE_NOTSET;
2655 16293 : data->pending_closed = false;
2656 16293 : data->seq = 0;
2657 :
2658 16293 : return DBRES_OK;
2659 : }
2660 :
2661 143 : void cloudsync_rollback_hook (void *ctx) {
2662 143 : cloudsync_context *data = (cloudsync_context *)ctx;
2663 :
2664 143 : data->pending_db_version = CLOUDSYNC_VALUE_NOTSET;
2665 143 : data->pending_closed = false;
2666 143 : data->seq = 0;
2667 143 : }
2668 :
2669 : // For a backend without commit and rollback hooks (PostgreSQL), called at the end of
2670 : // every transaction. It acts only on a transaction that took a db_version, so the cached
2671 : // db_version survives the many read-only transactions in between.
2672 0 : void cloudsync_transaction_end (cloudsync_context *data, bool committed) {
2673 0 : if (!data || data->pending_db_version == CLOUDSYNC_VALUE_NOTSET) return;
2674 0 : if (committed) cloudsync_commit_hook(data);
2675 0 : else cloudsync_rollback_hook(data);
2676 0 : }
2677 :
2678 : // Ends the pending db_version without committing it: the next change takes a new one with
2679 : // seq restarting at 0. The committed db_version is left alone, so a rollback restores it.
2680 8951 : static void cloudsync_pending_version_close (cloudsync_context *data) {
2681 8951 : if (data->pending_db_version == CLOUDSYNC_VALUE_NOTSET) return;
2682 351 : data->pending_closed = true;
2683 351 : data->seq = 0;
2684 8951 : }
2685 :
2686 24 : int cloudsync_begin_alter (cloudsync_context *data, const char *table_name) {
2687 : // init cloudsync_settings
2688 24 : if (cloudsync_context_init(data) == NULL) {
2689 0 : return DBRES_MISUSE;
2690 : }
2691 :
2692 : // lookup table
2693 24 : cloudsync_table_context *table = table_lookup(data, table_name);
2694 24 : if (!table) {
2695 : char buffer[1024];
2696 1 : snprintf(buffer, sizeof(buffer), "Unable to find table %s", table_name);
2697 1 : return cloudsync_set_error(data, buffer, DBRES_MISUSE);
2698 : }
2699 :
2700 : // idempotent: if already altering, return OK
2701 23 : if (table->is_altering) return DBRES_OK;
2702 :
2703 : // retrieve primary key(s)
2704 23 : char **names = NULL;
2705 23 : int nrows = 0;
2706 23 : int rc = database_pk_names(data, table_name, &names, &nrows);
2707 23 : if (rc != DBRES_OK) {
2708 : char buffer[1024];
2709 0 : snprintf(buffer, sizeof(buffer), "Unable to get primary keys for table %s", table_name);
2710 0 : cloudsync_set_error(data, buffer, DBRES_MISUSE);
2711 0 : goto rollback_begin_alter;
2712 : }
2713 :
2714 : // sanity check the number of primary keys
2715 23 : if (nrows != table_count_pks(table)) {
2716 : char buffer[1024];
2717 0 : snprintf(buffer, sizeof(buffer), "Number of primary keys for table %s changed before ALTER", table_name);
2718 0 : cloudsync_set_error(data, buffer, DBRES_MISUSE);
2719 0 : goto rollback_begin_alter;
2720 : }
2721 :
2722 : // drop original triggers
2723 23 : rc = database_delete_triggers(data, table_name);
2724 23 : if (rc != DBRES_OK) {
2725 : char buffer[1024];
2726 0 : snprintf(buffer, sizeof(buffer), "Unable to delete triggers for table %s in cloudsync_begin_alter.", table_name);
2727 0 : cloudsync_set_error(data, buffer, DBRES_ERROR);
2728 0 : goto rollback_begin_alter;
2729 : }
2730 :
2731 23 : table_set_pknames(table, names);
2732 23 : table->is_altering = true;
2733 23 : return DBRES_OK;
2734 :
2735 : rollback_begin_alter:
2736 0 : if (names) table_pknames_free(names, nrows);
2737 0 : return rc;
2738 24 : }
2739 :
2740 23 : int cloudsync_finalize_alter (cloudsync_context *data, cloudsync_table_context *table) {
2741 : // check if dbversion needed to be updated
2742 23 : cloudsync_dbversion_check_uptodate(data);
2743 :
2744 : // if primary-key columns change, all row identities change.
2745 : // In that case, the clock table must be dropped, recreated,
2746 : // and backfilled. We detect this by comparing the unique index
2747 : // in the lookaside table with the source table's PKs.
2748 :
2749 : // retrieve primary keys (to check is they changed)
2750 23 : char **result = NULL;
2751 23 : int nrows = 0;
2752 23 : int rc = database_pk_names (data, table->name, &result, &nrows);
2753 23 : if (rc != DBRES_OK || nrows == 0) {
2754 0 : if (nrows == 0) rc = DBRES_MISUSE;
2755 0 : goto finalize;
2756 : }
2757 :
2758 : // check if there are differences
2759 23 : bool pk_diff = (nrows != table->npks);
2760 23 : if (!pk_diff) {
2761 45 : for (int i = 0; i < nrows; ++i) {
2762 34 : if (strcmp(table->pk_name[i], result[i]) != 0) {
2763 6 : pk_diff = true;
2764 6 : break;
2765 : }
2766 28 : }
2767 17 : }
2768 :
2769 23 : if (pk_diff) {
2770 : // drop meta-table, it will be recreated
2771 12 : char *sql = cloudsync_memory_mprintf(SQL_DROP_CLOUDSYNC_TABLE, table->meta_ref);
2772 12 : rc = database_exec(data, sql);
2773 12 : cloudsync_memory_free(sql);
2774 12 : if (rc != DBRES_OK) {
2775 0 : DEBUG_DBERROR(rc, "cloudsync_finalize_alter", data);
2776 0 : goto finalize;
2777 : }
2778 12 : } else {
2779 : // compact meta-table
2780 : // delete entries for removed columns
2781 11 : const char *schema = table->schema ? table->schema : "";
2782 11 : char *sql = sql_build_delete_cols_not_in_schema_query(schema, table->name, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
2783 11 : rc = database_exec(data, sql);
2784 11 : cloudsync_memory_free(sql);
2785 11 : if (rc != DBRES_OK) {
2786 0 : DEBUG_DBERROR(rc, "cloudsync_finalize_alter", data);
2787 0 : goto finalize;
2788 : }
2789 :
2790 11 : sql = sql_build_pk_qualified_collist_query(schema, table->name);
2791 11 : if (!sql) {rc = DBRES_NOMEM; goto finalize;}
2792 :
2793 11 : char *pkclause = NULL;
2794 11 : rc = database_select_text(data, sql, &pkclause);
2795 11 : cloudsync_memory_free(sql);
2796 11 : if (rc != DBRES_OK) goto finalize;
2797 11 : char *pkvalues = (pkclause) ? pkclause : "rowid";
2798 :
2799 : // delete entries related to rows that no longer exist in the original table, but preserve tombstone
2800 11 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_GC_DELETE_ORPHANED_PK, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE, CLOUDSYNC_TOMBSTONE_VALUE, table->base_ref, table->meta_ref, pkvalues);
2801 11 : rc = database_exec(data, sql);
2802 11 : if (pkclause) cloudsync_memory_free(pkclause);
2803 11 : cloudsync_memory_free(sql);
2804 11 : if (rc != DBRES_OK) {
2805 0 : DEBUG_DBERROR(rc, "cloudsync_finalize_alter", data);
2806 0 : goto finalize;
2807 : }
2808 :
2809 : }
2810 :
2811 : // update key to be later used in cloudsync_dbversion_rebuild
2812 : char buf[256];
2813 23 : snprintf(buf, sizeof(buf), "%" PRId64, data->db_version);
2814 23 : dbutils_settings_set_key_value(data, "pre_alter_dbversion", buf);
2815 :
2816 : finalize:
2817 23 : table_pknames_free(result, nrows);
2818 23 : return rc;
2819 : }
2820 :
2821 24 : int cloudsync_commit_alter (cloudsync_context *data, const char *table_name) {
2822 24 : int rc = DBRES_MISUSE;
2823 24 : cloudsync_table_context *table = NULL;
2824 :
2825 : // init cloudsync_settings
2826 24 : if (cloudsync_context_init(data) == NULL) {
2827 0 : cloudsync_set_error(data, "Unable to initialize cloudsync context", DBRES_MISUSE);
2828 0 : goto rollback_finalize_alter;
2829 : }
2830 :
2831 : // lookup table
2832 24 : table = table_lookup(data, table_name);
2833 24 : if (!table) {
2834 : char buffer[1024];
2835 1 : snprintf(buffer, sizeof(buffer), "Unable to find table %s", table_name);
2836 1 : cloudsync_set_error(data, buffer, DBRES_MISUSE);
2837 1 : goto rollback_finalize_alter;
2838 : }
2839 :
2840 : // idempotent: if not altering, return OK
2841 23 : if (!table->is_altering) return DBRES_OK;
2842 :
2843 23 : rc = cloudsync_finalize_alter(data, table);
2844 23 : if (rc != DBRES_OK) goto rollback_finalize_alter;
2845 :
2846 : // the table is outdated, delete it and it will be reloaded in the cloudsync_init_internal
2847 : // is_altering is reset implicitly because table_free + cloudsync_init_table
2848 : // will reallocate the table context with zero-initialized memory
2849 23 : table_remove(data, table);
2850 23 : table_free(table);
2851 23 : table = NULL;
2852 :
2853 : // init again cloudsync for the table
2854 23 : table_algo algo_current = dbutils_table_settings_get_algo(data, table_name);
2855 23 : if (algo_current == table_algo_none) algo_current = dbutils_table_settings_get_algo(data, "*");
2856 23 : rc = cloudsync_init_table(data, table_name, cloudsync_algo_name(algo_current), CLOUDSYNC_INIT_FLAG_SKIP_INT_PK_CHECK);
2857 23 : if (rc != DBRES_OK) goto rollback_finalize_alter;
2858 :
2859 23 : return DBRES_OK;
2860 :
2861 : rollback_finalize_alter:
2862 1 : if (table) {
2863 0 : table_set_pknames(table, NULL);
2864 0 : table->is_altering = false;
2865 0 : }
2866 1 : return rc;
2867 24 : }
2868 :
2869 : // MARK: - Filter Rewrite -
2870 :
2871 : // Replace bare column names in a filter expression with prefix-qualified names.
2872 : // E.g., filter="user_id = 42", prefix="NEW", columns=["user_id","id"] → "NEW.\"user_id\" = 42"
2873 : // Columns must be sorted by length descending by the caller to avoid partial matches.
2874 : // Skips content inside single-quoted string literals.
2875 : // Returns a newly allocated string (caller must free with cloudsync_memory_free), or NULL on error.
2876 : // Helper: check if an identifier token matches a column name.
2877 112 : static bool filter_is_column (const char *token, size_t token_len, char **columns, int ncols) {
2878 450 : for (int i = 0; i < ncols; ++i) {
2879 388 : if (strlen(columns[i]) == token_len && strncmp(token, columns[i], token_len) == 0)
2880 50 : return true;
2881 338 : }
2882 62 : return false;
2883 112 : }
2884 :
2885 : // Helper: check if character is part of a SQL identifier.
2886 796 : static bool filter_is_ident_char (char c) {
2887 1262 : return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
2888 466 : (c >= '0' && c <= '9') || c == '_';
2889 : }
2890 :
2891 40 : char *cloudsync_filter_add_row_prefix (const char *filter, const char *prefix, char **columns, int ncols) {
2892 40 : if (!filter || !prefix || !columns || ncols <= 0) return NULL;
2893 :
2894 40 : size_t filter_len = strlen(filter);
2895 40 : size_t prefix_len = strlen(prefix);
2896 :
2897 : // Each identifier match grows by at most (prefix_len + 3) bytes.
2898 : // Worst case: the entire filter is one repeated column reference separated by
2899 : // single characters, so up to (filter_len / 2) matches. Use a safe upper bound.
2900 40 : size_t max_growth = (filter_len / 2 + 1) * (prefix_len + 3);
2901 40 : size_t cap = filter_len + max_growth + 64;
2902 40 : char *result = (char *)cloudsync_memory_alloc(cap);
2903 40 : if (!result) return NULL;
2904 40 : size_t out = 0;
2905 :
2906 : // Single pass: tokenize into identifiers, quoted strings, and everything else.
2907 40 : size_t i = 0;
2908 336 : while (i < filter_len) {
2909 : // Skip single-quoted string literals verbatim (handle '' escape)
2910 296 : if (filter[i] == '\'') {
2911 6 : result[out++] = filter[i++];
2912 38 : while (i < filter_len) {
2913 38 : if (filter[i] == '\'') {
2914 6 : result[out++] = filter[i++];
2915 : // '' is an escaped quote — keep going
2916 6 : if (i < filter_len && filter[i] == '\'') {
2917 0 : result[out++] = filter[i++];
2918 0 : continue;
2919 : }
2920 6 : break; // single ' ends the literal
2921 : }
2922 32 : result[out++] = filter[i++];
2923 : }
2924 6 : continue;
2925 : }
2926 :
2927 : // Extract identifier token
2928 290 : if (filter_is_ident_char(filter[i])) {
2929 112 : size_t start = i;
2930 540 : while (i < filter_len && filter_is_ident_char(filter[i])) ++i;
2931 112 : size_t token_len = i - start;
2932 :
2933 112 : if (filter_is_column(&filter[start], token_len, columns, ncols)) {
2934 : // Emit PREFIX."column_name"
2935 50 : memcpy(&result[out], prefix, prefix_len); out += prefix_len;
2936 50 : result[out++] = '.';
2937 50 : result[out++] = '"';
2938 50 : memcpy(&result[out], &filter[start], token_len); out += token_len;
2939 50 : result[out++] = '"';
2940 50 : } else {
2941 : // Not a column — copy as-is
2942 62 : memcpy(&result[out], &filter[start], token_len); out += token_len;
2943 : }
2944 112 : continue;
2945 : }
2946 :
2947 : // Any other character — copy as-is
2948 178 : result[out++] = filter[i++];
2949 : }
2950 :
2951 40 : result[out] = '\0';
2952 40 : return result;
2953 40 : }
2954 :
2955 24 : int cloudsync_reset_metatable (cloudsync_context *data, const char *table_name) {
2956 24 : cloudsync_table_context *table = table_lookup(data, table_name);
2957 24 : if (!table) return DBRES_ERROR;
2958 :
2959 24 : char *sql = cloudsync_memory_mprintf(SQL_DELETE_ALL_FROM_CLOUDSYNC_TABLE, table->meta_ref);
2960 24 : int rc = database_exec(data, sql);
2961 24 : cloudsync_memory_free(sql);
2962 24 : if (rc != DBRES_OK) return rc;
2963 :
2964 24 : return cloudsync_refill_metatable(data, table_name);
2965 24 : }
2966 :
2967 846 : int cloudsync_refill_metatable (cloudsync_context *data, const char *table_name) {
2968 846 : cloudsync_table_context *table = table_lookup(data, table_name);
2969 846 : if (!table) return DBRES_ERROR;
2970 :
2971 846 : dbvm_t *vm = NULL;
2972 846 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
2973 :
2974 : // Read row-level filter from settings (if any)
2975 : char filter_buf[2048];
2976 846 : int frc = dbutils_table_settings_get_value(data, table_name, "*", "filter", filter_buf, sizeof(filter_buf));
2977 846 : const char *filter = (frc == DBRES_OK && filter_buf[0]) ? filter_buf : NULL;
2978 :
2979 846 : const char *schema = table->schema ? table->schema : "";
2980 846 : char *sql = sql_build_pk_collist_query(schema, table_name);
2981 846 : char *pkclause_identifiers = NULL;
2982 846 : int rc = database_select_text(data, sql, &pkclause_identifiers);
2983 846 : cloudsync_memory_free(sql);
2984 846 : if (rc != DBRES_OK) goto finalize;
2985 846 : char *pkvalues_identifiers = (pkclause_identifiers) ? pkclause_identifiers : "rowid";
2986 :
2987 : // Use database-specific query builder to handle type differences in composite PKs
2988 846 : sql = sql_build_insert_missing_pks_query(schema, table_name, pkvalues_identifiers, table->base_ref, table->meta_ref, filter);
2989 846 : if (!sql) {rc = DBRES_NOMEM; goto finalize;}
2990 846 : rc = database_exec(data, sql);
2991 846 : cloudsync_memory_free(sql);
2992 846 : if (rc != DBRES_OK) goto finalize;
2993 :
2994 : // fill missing colums
2995 : // for each non-pk column:
2996 : // The new query does 1 encode per source row and one indexed NOT-EXISTS probe.
2997 : // The old plan does many decodes per candidate and can't use an index to rule out matches quickly—so it burns CPU and I/O.
2998 :
2999 846 : if (filter) {
3000 20 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_PKS_NOT_IN_SYNC_FOR_COL_FILTERED, pkvalues_identifiers, table->base_ref, filter, table->meta_ref);
3001 20 : } else {
3002 826 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_PKS_NOT_IN_SYNC_FOR_COL, pkvalues_identifiers, table->base_ref, table->meta_ref);
3003 : }
3004 846 : rc = databasevm_prepare(data, sql, (void **)&vm, DBFLAG_PERSISTENT);
3005 846 : cloudsync_memory_free(sql);
3006 846 : if (rc != DBRES_OK) goto finalize;
3007 :
3008 2801 : for (int i=0; i<table->ncols; ++i) {
3009 1956 : char *col_name = table->col_name[i];
3010 :
3011 1956 : rc = databasevm_bind_text(vm, 1, col_name, -1);
3012 1956 : if (rc != DBRES_OK) goto finalize;
3013 :
3014 1996 : while (1) {
3015 1996 : rc = databasevm_step(vm);
3016 1996 : if (rc == DBRES_ROW) {
3017 41 : size_t pklen = 0;
3018 41 : const void *pk = (const char *)database_column_blob(vm, 0, &pklen);
3019 41 : if (!pk) { rc = DBRES_ERROR; break; }
3020 41 : rc = local_mark_insert_or_update_meta(table, pk, pklen, col_name, db_version, cloudsync_bumpseq(data));
3021 41 : if (rc != DBRES_OK) break;
3022 1995 : } else if (rc == DBRES_DONE) {
3023 1955 : rc = DBRES_OK;
3024 1955 : break;
3025 : } else {
3026 0 : break;
3027 : }
3028 : }
3029 1956 : if (rc != DBRES_OK) goto finalize;
3030 :
3031 1955 : databasevm_reset(vm);
3032 2800 : }
3033 :
3034 : finalize:
3035 846 : if (rc != DBRES_OK) {DEBUG_ALWAYS("cloudsync_refill_metatable error: %s", database_errmsg(data));}
3036 846 : if (pkclause_identifiers) cloudsync_memory_free(pkclause_identifiers);
3037 846 : if (vm) databasevm_finalize(vm);
3038 846 : return rc;
3039 846 : }
3040 :
3041 : // MARK: - Local -
3042 :
3043 290 : int local_update_sentinel (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq) {
3044 290 : dbvm_t *vm = table->meta_sentinel_update_stmt;
3045 290 : if (!vm) return -1;
3046 :
3047 290 : int rc = databasevm_bind_int(vm, 1, db_version);
3048 290 : if (rc != DBRES_OK) goto cleanup;
3049 :
3050 290 : rc = databasevm_bind_int(vm, 2, seq);
3051 290 : if (rc != DBRES_OK) goto cleanup;
3052 :
3053 290 : rc = databasevm_bind_blob(vm, 3, pk, (int)pklen);
3054 290 : if (rc != DBRES_OK) goto cleanup;
3055 :
3056 290 : rc = databasevm_step(vm);
3057 290 : if (rc == DBRES_DONE) rc = DBRES_OK;
3058 :
3059 : cleanup:
3060 290 : DEBUG_DBERROR(rc, "local_update_sentinel", table->context);
3061 290 : databasevm_reset(vm);
3062 290 : return rc;
3063 290 : }
3064 :
3065 125 : int local_mark_insert_sentinel_meta (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq) {
3066 125 : dbvm_t *vm = table->meta_sentinel_insert_stmt;
3067 125 : if (!vm) return -1;
3068 :
3069 125 : int rc = databasevm_bind_blob(vm, 1, pk, (int)pklen);
3070 125 : if (rc != DBRES_OK) goto cleanup;
3071 :
3072 125 : rc = databasevm_bind_int(vm, 2, db_version);
3073 125 : if (rc != DBRES_OK) goto cleanup;
3074 :
3075 125 : rc = databasevm_bind_int(vm, 3, seq);
3076 125 : if (rc != DBRES_OK) goto cleanup;
3077 :
3078 125 : rc = databasevm_bind_int(vm, 4, db_version);
3079 125 : if (rc != DBRES_OK) goto cleanup;
3080 :
3081 125 : rc = databasevm_bind_int(vm, 5, seq);
3082 125 : if (rc != DBRES_OK) goto cleanup;
3083 :
3084 125 : rc = databasevm_step(vm);
3085 125 : if (rc == DBRES_DONE) rc = DBRES_OK;
3086 :
3087 : cleanup:
3088 125 : DEBUG_DBERROR(rc, "local_insert_sentinel", table->context);
3089 125 : databasevm_reset(vm);
3090 125 : return rc;
3091 125 : }
3092 :
3093 17036 : int local_mark_insert_or_update_meta_impl (cloudsync_table_context *table, const void *pk, size_t pklen, const char *col_name, int col_version, int64_t db_version, int seq) {
3094 :
3095 17036 : dbvm_t *vm = table->meta_row_insert_update_stmt;
3096 17036 : if (!vm) return -1;
3097 :
3098 17036 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
3099 17036 : if (rc != DBRES_OK) goto cleanup;
3100 :
3101 17036 : rc = databasevm_bind_text(vm, 2, (col_name) ? col_name : CLOUDSYNC_TOMBSTONE_VALUE, -1);
3102 17036 : if (rc != DBRES_OK) goto cleanup;
3103 :
3104 17036 : rc = databasevm_bind_int(vm, 3, col_version);
3105 17036 : if (rc != DBRES_OK) goto cleanup;
3106 :
3107 17036 : rc = databasevm_bind_int(vm, 4, db_version);
3108 17036 : if (rc != DBRES_OK) goto cleanup;
3109 :
3110 17036 : rc = databasevm_bind_int(vm, 5, seq);
3111 17036 : if (rc != DBRES_OK) goto cleanup;
3112 :
3113 17036 : rc = databasevm_bind_int(vm, 6, db_version);
3114 17036 : if (rc != DBRES_OK) goto cleanup;
3115 :
3116 17036 : rc = databasevm_bind_int(vm, 7, seq);
3117 17036 : if (rc != DBRES_OK) goto cleanup;
3118 :
3119 17036 : rc = databasevm_step(vm);
3120 17036 : if (rc == DBRES_DONE) rc = DBRES_OK;
3121 :
3122 : cleanup:
3123 17036 : DEBUG_DBERROR(rc, "local_insert_or_update", table->context);
3124 17036 : databasevm_reset(vm);
3125 17036 : return rc;
3126 17036 : }
3127 :
3128 16422 : int local_mark_insert_or_update_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const char *col_name, int64_t db_version, int seq) {
3129 16422 : return local_mark_insert_or_update_meta_impl(table, pk, pklen, col_name, 1, db_version, seq);
3130 : }
3131 :
3132 384 : int local_mark_delete_block_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const char *block_colname, int64_t db_version, int seq) {
3133 : // Mark a block as deleted by setting col_version = 2 (even = deleted)
3134 384 : return local_mark_insert_or_update_meta_impl(table, pk, pklen, block_colname, 2, db_version, seq);
3135 : }
3136 :
3137 384 : int block_delete_value_external (cloudsync_context *data, cloudsync_table_context *table, const void *pk, size_t pklen, const char *block_colname) {
3138 384 : return block_delete_value(data, table, pk, (int)pklen, block_colname);
3139 : }
3140 :
3141 230 : int local_mark_delete_meta (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq) {
3142 230 : return local_mark_insert_or_update_meta_impl(table, pk, pklen, NULL, 2, db_version, seq);
3143 : }
3144 :
3145 202 : int local_drop_meta (cloudsync_table_context *table, const void *pk, size_t pklen) {
3146 202 : dbvm_t *vm = table->meta_row_drop_stmt;
3147 202 : if (!vm) return -1;
3148 :
3149 202 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
3150 202 : if (rc != DBRES_OK) goto cleanup;
3151 :
3152 202 : rc = databasevm_step(vm);
3153 202 : if (rc == DBRES_DONE) rc = DBRES_OK;
3154 :
3155 : cleanup:
3156 202 : DEBUG_DBERROR(rc, "local_drop_meta", table->context);
3157 202 : databasevm_reset(vm);
3158 202 : return rc;
3159 202 : }
3160 :
3161 28 : int local_update_move_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const void *pk2, size_t pklen2, int64_t db_version) {
3162 : /*
3163 : * This function moves non-sentinel metadata entries from an old primary key (OLD.pk)
3164 : * to a new primary key (NEW.pk) when a primary key change occurs.
3165 : *
3166 : * To ensure consistency and proper conflict resolution in a CRDT (Conflict-free Replicated Data Type) system,
3167 : * each non-sentinel metadata entry involved in the move must have a unique sequence value (seq).
3168 : *
3169 : * The `seq` is crucial for tracking the order of operations and for detecting and resolving conflicts
3170 : * during synchronization between replicas. Without a unique `seq` for each entry, concurrent updates
3171 : * may be applied incorrectly, leading to data inconsistency.
3172 : *
3173 : * When performing the update, a unique `seq` must be assigned to each metadata row. This can be achieved
3174 : * by either incrementing the maximum sequence value in the table or using a function (e.g., cloudsync_bumpseq(data))
3175 : * that generates a unique sequence for each row. The update query should ensure that each row moved
3176 : * from OLD.pk to NEW.pk gets a distinct `seq` to maintain proper versioning and ordering of changes.
3177 : */
3178 :
3179 : // see https://github.com/sqliteai/sqlite-sync/blob/main/docs/PriKey.md for more details
3180 : // pk2 is the old pk
3181 :
3182 28 : dbvm_t *vm = table->meta_update_move_stmt;
3183 28 : if (!vm) return -1;
3184 :
3185 : // new primary key
3186 28 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
3187 28 : if (rc != DBRES_OK) goto cleanup;
3188 :
3189 : // new db_version
3190 28 : rc = databasevm_bind_int(vm, 2, db_version);
3191 28 : if (rc != DBRES_OK) goto cleanup;
3192 :
3193 : // old primary key
3194 28 : rc = databasevm_bind_blob(vm, 3, pk2, pklen2);
3195 28 : if (rc != DBRES_OK) goto cleanup;
3196 :
3197 28 : rc = databasevm_step(vm);
3198 28 : if (rc == DBRES_DONE) rc = DBRES_OK;
3199 :
3200 : cleanup:
3201 28 : DEBUG_DBERROR(rc, "local_update_move_meta", table->context);
3202 28 : databasevm_reset(vm);
3203 28 : return rc;
3204 28 : }
3205 :
3206 : // MARK: - Payload Encode / Decode -
3207 :
3208 2129 : static void cloudsync_payload_checksum_store (cloudsync_payload_header *header, uint64_t checksum) {
3209 2129 : uint64_t h = checksum & 0xFFFFFFFFFFFFULL; // keep 48 bits
3210 2129 : header->checksum[0] = (uint8_t)(h >> 40);
3211 2129 : header->checksum[1] = (uint8_t)(h >> 32);
3212 2129 : header->checksum[2] = (uint8_t)(h >> 24);
3213 2129 : header->checksum[3] = (uint8_t)(h >> 16);
3214 2129 : header->checksum[4] = (uint8_t)(h >> 8);
3215 2129 : header->checksum[5] = (uint8_t)(h >> 0);
3216 2129 : }
3217 :
3218 1916 : static uint64_t cloudsync_payload_checksum_load (cloudsync_payload_header *header) {
3219 5748 : return ((uint64_t)header->checksum[0] << 40) |
3220 3832 : ((uint64_t)header->checksum[1] << 32) |
3221 3832 : ((uint64_t)header->checksum[2] << 24) |
3222 3832 : ((uint64_t)header->checksum[3] << 16) |
3223 3832 : ((uint64_t)header->checksum[4] << 8) |
3224 1916 : ((uint64_t)header->checksum[5] << 0);
3225 : }
3226 :
3227 1916 : static bool cloudsync_payload_checksum_verify (cloudsync_payload_header *header, uint64_t checksum) {
3228 1916 : uint64_t checksum1 = cloudsync_payload_checksum_load(header);
3229 1916 : uint64_t checksum2 = checksum & 0xFFFFFFFFFFFFULL;
3230 1916 : return (checksum1 == checksum2);
3231 : }
3232 :
3233 60556 : static bool cloudsync_payload_encode_check (cloudsync_payload_context *payload, size_t needed) {
3234 60556 : if (payload->nrows == 0) needed += sizeof(cloudsync_payload_header);
3235 :
3236 : // alloc/resize buffer
3237 60556 : if (payload->bused + needed > payload->balloc) {
3238 2149 : if (needed < CLOUDSYNC_PAYLOAD_MINBUF_SIZE) needed = CLOUDSYNC_PAYLOAD_MINBUF_SIZE;
3239 2149 : size_t balloc = payload->balloc + needed;
3240 :
3241 2149 : char *buffer = cloudsync_memory_realloc(payload->buffer, balloc);
3242 2149 : if (!buffer) {
3243 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3244 0 : memset(payload, 0, sizeof(cloudsync_payload_context));
3245 0 : return false;
3246 : }
3247 :
3248 2149 : payload->buffer = buffer;
3249 2149 : payload->balloc = balloc;
3250 2149 : if (payload->nrows == 0) payload->bused = sizeof(cloudsync_payload_header);
3251 2149 : }
3252 :
3253 60556 : return true;
3254 60556 : }
3255 :
3256 61034 : size_t cloudsync_payload_context_size (size_t *header_size) {
3257 61034 : if (header_size) *header_size = sizeof(cloudsync_payload_header);
3258 61034 : return sizeof(cloudsync_payload_context);
3259 : }
3260 :
3261 0 : void cloudsync_payload_context_free (cloudsync_payload_context *payload) {
3262 0 : if (!payload) return;
3263 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3264 0 : cloudsync_memory_free(payload);
3265 0 : }
3266 :
3267 7527 : uint64_t cloudsync_payload_context_nrows (cloudsync_payload_context *payload) {
3268 7527 : return payload ? payload->nrows : 0;
3269 : }
3270 :
3271 3595 : size_t cloudsync_payload_context_bused (cloudsync_payload_context *payload) {
3272 3595 : return payload ? payload->bused : 0;
3273 : }
3274 :
3275 2129 : void cloudsync_payload_header_init (cloudsync_payload_header *header, uint8_t version, uint32_t expanded_size, uint16_t ncols, uint32_t nrows, uint64_t hash) {
3276 2129 : memset(header, 0, sizeof(cloudsync_payload_header));
3277 : assert(sizeof(cloudsync_payload_header)==32);
3278 :
3279 : int major, minor, patch;
3280 2129 : sscanf(CLOUDSYNC_VERSION, "%d.%d.%d", &major, &minor, &patch);
3281 :
3282 2129 : header->signature = htonl(CLOUDSYNC_PAYLOAD_SIGNATURE);
3283 2129 : header->version = version;
3284 2129 : header->libversion[0] = (uint8_t)major;
3285 2129 : header->libversion[1] = (uint8_t)minor;
3286 2129 : header->libversion[2] = (uint8_t)patch;
3287 2129 : header->expanded_size = htonl(expanded_size);
3288 2129 : header->ncols = htons(ncols);
3289 2129 : header->nrows = htonl(nrows);
3290 2129 : header->schema_hash = htonll(hash);
3291 2129 : }
3292 :
3293 60423 : int cloudsync_payload_encode_step (cloudsync_payload_context *payload, cloudsync_context *data, int argc, dbvalue_t **argv) {
3294 : DEBUG_FUNCTION("cloudsync_payload_encode_step");
3295 : // debug_values(argc, argv);
3296 :
3297 : // check if the step function is called for the first time
3298 60423 : if (payload->nrows == 0) payload->ncols = (uint16_t)argc;
3299 :
3300 60423 : size_t breq = pk_encode_size((dbvalue_t **)argv, argc, 0, data->skip_decode_idx);
3301 60423 : if (cloudsync_payload_encode_check(payload, breq) == false) {
3302 0 : return cloudsync_set_error(data, "Not enough memory to resize payload internal buffer", DBRES_NOMEM);
3303 : }
3304 :
3305 60423 : char *buffer = payload->buffer + payload->bused;
3306 60423 : size_t bsize = payload->balloc - payload->bused;
3307 60423 : char *p = pk_encode((dbvalue_t **)argv, argc, buffer, false, &bsize, data->skip_decode_idx);
3308 60423 : if (!p) return cloudsync_set_error(data, "An error occurred while encoding payload", DBRES_ERROR);
3309 :
3310 : // update buffer
3311 60423 : payload->bused += breq;
3312 :
3313 : // increment row counter
3314 60423 : ++payload->nrows;
3315 :
3316 60423 : return DBRES_OK;
3317 60423 : }
3318 :
3319 133 : static bool cloudsync_payload_append_raw (cloudsync_payload_context *payload, cloudsync_context *data, const char **fields, const size_t *field_sizes, int nfields, uint8_t version) {
3320 133 : size_t needed = 0;
3321 1330 : for (int i = 0; i < nfields; ++i) {
3322 1197 : if (field_sizes[i] > SIZE_MAX - needed) {
3323 0 : cloudsync_set_error(data, CLOUDSYNC_ERRCODE_ROW_TOO_LARGE "cloudsync payload raw row too large", DBRES_NOMEM);
3324 0 : return false;
3325 : }
3326 1197 : needed += field_sizes[i];
3327 1197 : }
3328 133 : if (!cloudsync_payload_encode_check(payload, needed)) {
3329 0 : cloudsync_set_error(data, "Not enough memory to resize payload internal buffer", DBRES_NOMEM);
3330 0 : return false;
3331 : }
3332 133 : if (payload->nrows == 0) {
3333 133 : payload->ncols = (uint16_t)nfields;
3334 133 : payload->version = version;
3335 133 : }
3336 133 : char *dst = payload->buffer + payload->bused;
3337 1330 : for (int i = 0; i < nfields; ++i) {
3338 1197 : memcpy(dst, fields[i], field_sizes[i]);
3339 1197 : dst += field_sizes[i];
3340 1197 : }
3341 133 : payload->bused += needed;
3342 133 : ++payload->nrows;
3343 133 : return true;
3344 133 : }
3345 :
3346 368 : int cloudsync_payload_max_chunk_size (cloudsync_context *data) {
3347 368 : int64_t value = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_PAYLOAD_MAX_CHUNK_SIZE);
3348 368 : if (value <= 0) value = CLOUDSYNC_PAYLOAD_CHUNK_DEFAULT_SIZE;
3349 368 : if (value < CLOUDSYNC_PAYLOAD_CHUNK_MIN_SIZE) value = CLOUDSYNC_PAYLOAD_CHUNK_MIN_SIZE;
3350 368 : if (value > CLOUDSYNC_PAYLOAD_CHUNK_MAX_SIZE) value = CLOUDSYNC_PAYLOAD_CHUNK_MAX_SIZE;
3351 368 : return (int)value;
3352 : }
3353 :
3354 0 : int cloudsync_payload_fragment_target_size (cloudsync_context *data) {
3355 0 : int max_size = cloudsync_payload_max_chunk_size(data);
3356 0 : int target = max_size - (int)sizeof(cloudsync_payload_header) - CLOUDSYNC_PAYLOAD_CHUNK_SAFETY_MARGIN;
3357 0 : if (target < 1024) target = 1024;
3358 0 : return target;
3359 : }
3360 :
3361 366 : static size_t cloudsync_payload_decimal_len_i64 (int64_t value) {
3362 366 : size_t len = value < 0 ? 1 : 0;
3363 366 : uint64_t v = (value < 0) ? (uint64_t)(-(value + 1)) + 1u : (uint64_t)value;
3364 366 : do {
3365 992 : len++;
3366 992 : v /= 10u;
3367 992 : } while (v != 0);
3368 366 : return len;
3369 : }
3370 :
3371 976 : static bool cloudsync_payload_size_add (size_t *acc, size_t value) {
3372 976 : if (value > SIZE_MAX - *acc) return false;
3373 976 : *acc += value;
3374 976 : return true;
3375 976 : }
3376 :
3377 122 : int cloudsync_payload_fragment_count (int64_t total_size, int target_size) {
3378 122 : if (total_size <= 0 || target_size <= 0) return 0;
3379 122 : uint64_t total = (uint64_t)total_size;
3380 122 : uint64_t target = (uint64_t)target_size;
3381 122 : uint64_t count = total / target + ((total % target) != 0);
3382 122 : if (count == 0 || count > INT_MAX) return 0;
3383 122 : return (int)count;
3384 122 : }
3385 :
3386 122 : int cloudsync_payload_fragment_data_size (cloudsync_context *data,
3387 : const char *tbl, int tbl_len,
3388 : const void *pk, int pk_len,
3389 : const char *col_name, int col_name_len,
3390 : int64_t col_version, int64_t db_version,
3391 : const void *site_id, int site_id_len,
3392 : int64_t cl, int64_t seq,
3393 : int64_t total_size,
3394 : int part_index, int part_count) {
3395 122 : UNUSED_PARAMETER(pk);
3396 122 : UNUSED_PARAMETER(site_id);
3397 122 : if (tbl_len < 0 && tbl) tbl_len = (int)strlen(tbl);
3398 122 : if (col_name_len < 0 && col_name) col_name_len = (int)strlen(col_name);
3399 122 : if (tbl_len < 0 || pk_len < 0 || col_name_len < 0 || site_id_len < 0 || total_size < 0 || part_index < 0 || part_count <= 0) {
3400 0 : return 0;
3401 : }
3402 :
3403 122 : size_t fixed = sizeof(cloudsync_payload_header);
3404 244 : size_t frag_col_len = strlen(CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX) + 32 + 1 + 16 + 1 +
3405 366 : cloudsync_payload_decimal_len_i64(part_index) + 1 +
3406 366 : cloudsync_payload_decimal_len_i64(part_count) + 1 +
3407 366 : cloudsync_payload_decimal_len_i64(total_size) + 1 +
3408 122 : (size_t)col_name_len;
3409 976 : size_t sizes[] = {
3410 122 : pk_encode_raw_size(DBTYPE_TEXT, tbl_len),
3411 122 : pk_encode_raw_size(DBTYPE_BLOB, pk_len),
3412 122 : pk_encode_raw_size(DBTYPE_TEXT, (int64_t)frag_col_len),
3413 122 : pk_encode_raw_size(DBTYPE_INTEGER, col_version),
3414 122 : pk_encode_raw_size(DBTYPE_INTEGER, db_version),
3415 122 : pk_encode_raw_size(DBTYPE_BLOB, site_id_len),
3416 122 : pk_encode_raw_size(DBTYPE_INTEGER, cl),
3417 122 : pk_encode_raw_size(DBTYPE_INTEGER, seq)
3418 : };
3419 1098 : for (size_t i = 0; i < sizeof(sizes) / sizeof(sizes[0]); ++i) {
3420 976 : if (sizes[i] == SIZE_MAX || !cloudsync_payload_size_add(&fixed, sizes[i])) return 0;
3421 976 : }
3422 :
3423 122 : int max_size = cloudsync_payload_max_chunk_size(data);
3424 122 : if (fixed >= (size_t)max_size) return 0;
3425 :
3426 122 : size_t candidate = (size_t)max_size - fixed;
3427 122 : if (candidate > INT_MAX) candidate = INT_MAX;
3428 244 : while (candidate > 0) {
3429 244 : size_t frag_size = pk_encode_raw_size(DBTYPE_BLOB, (int64_t)candidate);
3430 244 : if (frag_size == SIZE_MAX) return 0;
3431 244 : if (fixed <= (size_t)max_size && frag_size <= (size_t)max_size - fixed) return (int)candidate;
3432 122 : size_t total = fixed + frag_size;
3433 122 : size_t over = total > (size_t)max_size ? total - (size_t)max_size : 1;
3434 122 : if (candidate <= over) return 0;
3435 122 : candidate -= over;
3436 : }
3437 0 : return 0;
3438 122 : }
3439 :
3440 122 : int cloudsync_payload_encoded_value_header (dbvalue_t *value, char *header, int header_cap, int64_t *payload_len) {
3441 122 : if (!value || !header || header_cap <= 0 || !payload_len) return -1;
3442 122 : int type = database_value_type(value);
3443 122 : *payload_len = 0;
3444 122 : if (type != DBTYPE_TEXT && type != DBTYPE_BLOB) return 0;
3445 122 : int64_t len = database_value_bytes(value);
3446 122 : if (len < 0) return -1;
3447 122 : *payload_len = len;
3448 122 : size_t total = pk_encode_raw_size(type, len);
3449 122 : if (total == SIZE_MAX || total < (size_t)len || total - (size_t)len > (size_t)header_cap) return -1;
3450 122 : if (type == DBTYPE_TEXT) {
3451 18 : size_t nbytes = pk_encode_raw_size(type, len) - (size_t)len - 1;
3452 18 : uint8_t type_byte = (uint8_t)((nbytes << 3) | DBTYPE_TEXT);
3453 18 : header[0] = (char)type_byte;
3454 72 : for (size_t i = 0; i < nbytes; i++) header[1 + i] = (uint8_t)(((uint64_t)len >> (8 * (nbytes - 1 - i))) & 0xFFu);
3455 18 : return (int)(1 + nbytes);
3456 : } else {
3457 104 : size_t nbytes = pk_encode_raw_size(type, len) - (size_t)len - 1;
3458 104 : uint8_t type_byte = (uint8_t)((nbytes << 3) | DBTYPE_BLOB);
3459 104 : header[0] = (char)type_byte;
3460 416 : for (size_t i = 0; i < nbytes; i++) header[1 + i] = (uint8_t)(((uint64_t)len >> (8 * (nbytes - 1 - i))) & 0xFFu);
3461 104 : return (int)(1 + nbytes);
3462 : }
3463 122 : }
3464 :
3465 61 : uint64_t cloudsync_payload_encoded_value_checksum (dbvalue_t *value) {
3466 61 : if (!value) return 0;
3467 61 : int type = database_value_type(value);
3468 61 : if (type != DBTYPE_TEXT && type != DBTYPE_BLOB) {
3469 0 : size_t len = pk_encode_size(&value, 1, 0, -1);
3470 : char stack[32];
3471 0 : char *buf = stack;
3472 0 : if (len > sizeof(stack)) buf = cloudsync_memory_alloc((uint64_t)len);
3473 0 : if (!buf) return 0;
3474 0 : size_t bsize = len;
3475 0 : pk_encode(&value, 1, buf, false, &bsize, -1);
3476 0 : uint64_t h = pk_checksum(buf, bsize);
3477 0 : if (buf != stack) cloudsync_memory_free(buf);
3478 0 : return h;
3479 : }
3480 : char header[16];
3481 61 : int64_t payload_len = 0;
3482 61 : int header_len = cloudsync_payload_encoded_value_header(value, header, sizeof(header), &payload_len);
3483 61 : if (header_len <= 0) return 0;
3484 61 : uint64_t h = pk_checksum(header, (size_t)header_len);
3485 61 : const char *p = (const char *)database_value_blob(value);
3486 61 : if (p && payload_len > 0) {
3487 61 : const uint8_t *bytes = (const uint8_t *)p;
3488 35620061 : for (int64_t i = 0; i < payload_len; ++i) {
3489 35620000 : h ^= bytes[i];
3490 35620000 : h *= 1099511628211ULL;
3491 35620000 : }
3492 61 : }
3493 61 : return h;
3494 61 : }
3495 :
3496 23078 : static uint64_t cloudsync_checksum_update (uint64_t h, const void *data, size_t len) {
3497 23078 : const uint8_t *p = (const uint8_t *)data;
3498 17456887 : for (size_t i = 0; i < len; ++i) {
3499 17433809 : h ^= p[i];
3500 17433809 : h *= 1099511628211ULL;
3501 17433809 : }
3502 23078 : return h;
3503 : }
3504 :
3505 2580 : static uint64_t cloudsync_checksum_update_i64 (uint64_t h, int64_t value) {
3506 2580 : uint64_t v = (uint64_t)value;
3507 23220 : for (int i = 7; i >= 0; --i) {
3508 20640 : uint8_t b = (uint8_t)((v >> (8 * i)) & 0xffu);
3509 20640 : h = cloudsync_checksum_update(h, &b, 1);
3510 20640 : }
3511 2580 : return h;
3512 : }
3513 :
3514 215 : static void cloudsync_payload_fragment_value_id (char out[33],
3515 : const char *tbl, int tbl_len,
3516 : const void *pk, int pk_len,
3517 : const char *col_name, int col_name_len,
3518 : int64_t col_version, int64_t db_version,
3519 : const void *site_id, int site_id_len,
3520 : int64_t cl, int64_t seq,
3521 : uint64_t value_checksum,
3522 : int64_t total_size) {
3523 215 : uint64_t h1 = 14695981039346656037ULL;
3524 215 : uint64_t h2 = 1099511628211ULL;
3525 215 : const char sep = '\x1f';
3526 :
3527 215 : h1 = cloudsync_checksum_update(h1, tbl, (size_t)tbl_len);
3528 215 : h1 = cloudsync_checksum_update(h1, &sep, 1);
3529 215 : h1 = cloudsync_checksum_update(h1, pk, (size_t)pk_len);
3530 215 : h1 = cloudsync_checksum_update(h1, &sep, 1);
3531 215 : h1 = cloudsync_checksum_update(h1, col_name, (size_t)col_name_len);
3532 215 : h1 = cloudsync_checksum_update(h1, &sep, 1);
3533 215 : h1 = cloudsync_checksum_update(h1, site_id, (size_t)site_id_len);
3534 215 : h1 = cloudsync_checksum_update_i64(h1, col_version);
3535 215 : h1 = cloudsync_checksum_update_i64(h1, db_version);
3536 215 : h1 = cloudsync_checksum_update_i64(h1, cl);
3537 215 : h1 = cloudsync_checksum_update_i64(h1, seq);
3538 215 : h1 = cloudsync_checksum_update_i64(h1, (int64_t)value_checksum);
3539 215 : h1 = cloudsync_checksum_update_i64(h1, total_size);
3540 :
3541 215 : h2 = cloudsync_checksum_update_i64(h2, total_size);
3542 215 : h2 = cloudsync_checksum_update_i64(h2, (int64_t)value_checksum);
3543 215 : h2 = cloudsync_checksum_update(h2, site_id, (size_t)site_id_len);
3544 215 : h2 = cloudsync_checksum_update(h2, col_name, (size_t)col_name_len);
3545 215 : h2 = cloudsync_checksum_update(h2, pk, (size_t)pk_len);
3546 215 : h2 = cloudsync_checksum_update(h2, tbl, (size_t)tbl_len);
3547 215 : h2 = cloudsync_checksum_update_i64(h2, seq);
3548 215 : h2 = cloudsync_checksum_update_i64(h2, cl);
3549 215 : h2 = cloudsync_checksum_update_i64(h2, db_version);
3550 215 : h2 = cloudsync_checksum_update_i64(h2, col_version);
3551 :
3552 215 : snprintf(out, 33, "%016" PRIx64 "%016" PRIx64, h1, h2);
3553 215 : }
3554 :
3555 133 : int cloudsync_payload_encode_fragment_step (cloudsync_payload_context *payload, cloudsync_context *data,
3556 : const char *tbl, int tbl_len,
3557 : const void *pk, int pk_len,
3558 : const char *col_name, int col_name_len,
3559 : const void *fragment, int fragment_len,
3560 : int64_t col_version, int64_t db_version,
3561 : const void *site_id, int site_id_len,
3562 : int64_t cl, int64_t seq,
3563 : uint64_t value_checksum,
3564 : int64_t total_size,
3565 : int part_index, int part_count) {
3566 133 : if (!payload || !data || !tbl || !pk || !col_name || !fragment || !site_id) return DBRES_MISUSE;
3567 133 : if (tbl_len < 0) tbl_len = (int)strlen(tbl);
3568 133 : if (col_name_len < 0) col_name_len = (int)strlen(col_name);
3569 266 : if (tbl_len < 0 || pk_len < 0 || col_name_len < 0 || fragment_len <= 0 || site_id_len < 0 ||
3570 133 : total_size <= 0 || part_index < 0 || part_count <= 0 || part_index >= part_count) {
3571 0 : return DBRES_MISUSE;
3572 : }
3573 :
3574 : char value_id[33];
3575 : char checksum_hex[17];
3576 266 : cloudsync_payload_fragment_value_id(value_id, tbl, tbl_len, pk, pk_len, col_name, col_name_len,
3577 133 : col_version, db_version, site_id, site_id_len, cl, seq,
3578 133 : value_checksum, total_size);
3579 133 : snprintf(checksum_hex, sizeof(checksum_hex), "%016" PRIx64, value_checksum);
3580 :
3581 133 : char *frag_col = cloudsync_memory_mprintf("%s%s:%s:%d:%d:%" PRId64 ":%.*s",
3582 : CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX,
3583 133 : value_id, checksum_hex, part_index, part_count, total_size,
3584 133 : col_name_len, col_name);
3585 133 : if (!frag_col) return DBRES_NOMEM;
3586 :
3587 133 : size_t sizes[9] = {0};
3588 133 : sizes[0] = pk_encode_raw_size(DBTYPE_TEXT, tbl_len);
3589 133 : sizes[1] = pk_encode_raw_size(DBTYPE_BLOB, pk_len);
3590 133 : sizes[2] = pk_encode_raw_size(DBTYPE_TEXT, (int64_t)strlen(frag_col));
3591 133 : sizes[3] = pk_encode_raw_size(DBTYPE_BLOB, fragment_len);
3592 133 : sizes[4] = pk_encode_raw_size(DBTYPE_INTEGER, col_version);
3593 133 : sizes[5] = pk_encode_raw_size(DBTYPE_INTEGER, db_version);
3594 133 : sizes[6] = pk_encode_raw_size(DBTYPE_BLOB, site_id_len);
3595 133 : sizes[7] = pk_encode_raw_size(DBTYPE_INTEGER, cl);
3596 133 : sizes[8] = pk_encode_raw_size(DBTYPE_INTEGER, seq);
3597 1330 : for (int i = 0; i < 9; ++i) {
3598 1197 : if (sizes[i] == SIZE_MAX) { cloudsync_memory_free(frag_col); return DBRES_NOMEM; }
3599 1197 : }
3600 :
3601 : char stack[9][64];
3602 133 : char *fields[9] = {0};
3603 1330 : for (int i = 0; i < 9; ++i) {
3604 1197 : fields[i] = sizes[i] <= sizeof(stack[0]) ? stack[i] : cloudsync_memory_alloc((uint64_t)sizes[i]);
3605 1197 : if (!fields[i]) {
3606 0 : for (int j = 0; j < i; ++j) if (fields[j] && (fields[j] < (char *)stack || fields[j] >= (char *)(stack + 9))) cloudsync_memory_free(fields[j]);
3607 0 : cloudsync_memory_free(frag_col);
3608 0 : return DBRES_NOMEM;
3609 : }
3610 1197 : }
3611 :
3612 133 : pk_encode_raw_text(fields[0], tbl, (size_t)tbl_len);
3613 133 : pk_encode_raw_blob(fields[1], pk, (size_t)pk_len);
3614 133 : pk_encode_raw_text(fields[2], frag_col, strlen(frag_col));
3615 133 : pk_encode_raw_blob(fields[3], fragment, (size_t)fragment_len);
3616 133 : pk_encode_raw_int(fields[4], col_version);
3617 133 : pk_encode_raw_int(fields[5], db_version);
3618 133 : pk_encode_raw_blob(fields[6], site_id, (size_t)site_id_len);
3619 133 : pk_encode_raw_int(fields[7], cl);
3620 133 : pk_encode_raw_int(fields[8], seq);
3621 :
3622 : const char *cfields[9];
3623 1330 : for (int i = 0; i < 9; ++i) cfields[i] = fields[i];
3624 133 : bool ok = cloudsync_payload_append_raw(payload, data, cfields, sizes, 9, CLOUDSYNC_PAYLOAD_VERSION_3);
3625 :
3626 1330 : for (int i = 0; i < 9; ++i) {
3627 1197 : if (!(fields[i] >= (char *)stack && fields[i] < (char *)(stack + 9))) cloudsync_memory_free(fields[i]);
3628 1197 : }
3629 133 : cloudsync_memory_free(frag_col);
3630 133 : return ok ? DBRES_OK : cloudsync_errcode(data);
3631 133 : }
3632 :
3633 2137 : int cloudsync_payload_encode_final (cloudsync_payload_context *payload, cloudsync_context *data) {
3634 : DEBUG_FUNCTION("cloudsync_payload_encode_final");
3635 :
3636 2137 : if (payload->nrows == 0) {
3637 8 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3638 8 : payload->buffer = NULL;
3639 8 : payload->bsize = 0;
3640 8 : return DBRES_OK;
3641 : }
3642 :
3643 2129 : if (payload->nrows > UINT32_MAX) {
3644 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3645 0 : payload->buffer = NULL;
3646 0 : payload->bsize = 0;
3647 0 : cloudsync_set_error(data, "Maximum number of payload rows reached", DBRES_ERROR);
3648 0 : return DBRES_ERROR;
3649 : }
3650 :
3651 : // sanity check about buffer size
3652 2129 : int header_size = (int)sizeof(cloudsync_payload_header);
3653 2129 : int64_t buffer_size = (int64_t)payload->bused - (int64_t)header_size;
3654 2129 : if (buffer_size < 0) {
3655 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3656 0 : payload->buffer = NULL;
3657 0 : payload->bsize = 0;
3658 0 : cloudsync_set_error(data, "cloudsync_encode: internal size underflow", DBRES_ERROR);
3659 0 : return DBRES_ERROR;
3660 : }
3661 2129 : if (buffer_size > INT_MAX) {
3662 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3663 0 : payload->buffer = NULL;
3664 0 : payload->bsize = 0;
3665 0 : cloudsync_set_error(data, CLOUDSYNC_ERRCODE_PAYLOAD_TOO_LARGE "cloudsync_encode: payload too large to compress (INT_MAX limit)", DBRES_ERROR);
3666 0 : return DBRES_ERROR;
3667 : }
3668 : // try to allocate buffer used for compressed data
3669 2129 : int real_buffer_size = (int)buffer_size;
3670 2129 : int zbound = LZ4_compressBound(real_buffer_size);
3671 2129 : char *zbuffer = cloudsync_memory_alloc(zbound + header_size); // if for some reasons allocation fails then just skip compression
3672 :
3673 : // skip the reserved header from the buffer to compress
3674 2129 : char *src_buffer = payload->buffer + sizeof(cloudsync_payload_header);
3675 2129 : int zused = (zbuffer) ? LZ4_compress_default(src_buffer, zbuffer+header_size, real_buffer_size, zbound) : 0;
3676 2129 : bool use_uncompressed_buffer = (!zused || zused > real_buffer_size);
3677 2129 : CHECK_FORCE_UNCOMPRESSED_BUFFER();
3678 :
3679 : // setup payload header
3680 2129 : cloudsync_payload_header header = {0};
3681 2129 : uint32_t expanded_size = (use_uncompressed_buffer) ? 0 : real_buffer_size;
3682 2129 : uint8_t version = payload->version ? payload->version : CLOUDSYNC_PAYLOAD_VERSION_LATEST;
3683 2129 : cloudsync_payload_header_init(&header, version, expanded_size, payload->ncols, (uint32_t)payload->nrows, data->schema_hash);
3684 :
3685 : // if compression fails or if compressed size is bigger than original buffer, then use the uncompressed buffer
3686 2129 : if (use_uncompressed_buffer) {
3687 400 : if (zbuffer) cloudsync_memory_free(zbuffer);
3688 400 : zbuffer = payload->buffer;
3689 400 : zused = real_buffer_size;
3690 400 : }
3691 :
3692 : // compute checksum of the buffer
3693 2129 : uint64_t checksum = pk_checksum(zbuffer + header_size, zused);
3694 2129 : cloudsync_payload_checksum_store(&header, checksum);
3695 :
3696 : // copy header and data to SQLite BLOB
3697 2129 : memcpy(zbuffer, &header, sizeof(cloudsync_payload_header));
3698 2129 : int blob_size = zused + sizeof(cloudsync_payload_header);
3699 2129 : payload->bsize = blob_size;
3700 :
3701 : // cleanup memory
3702 2129 : if (zbuffer != payload->buffer) {
3703 1729 : cloudsync_memory_free (payload->buffer);
3704 1729 : payload->buffer = zbuffer;
3705 1729 : }
3706 :
3707 2129 : return DBRES_OK;
3708 2137 : }
3709 :
3710 2137 : char *cloudsync_payload_blob (cloudsync_payload_context *payload, int64_t *blob_size, int64_t *nrows) {
3711 : DEBUG_FUNCTION("cloudsync_payload_blob");
3712 :
3713 2137 : if (blob_size) *blob_size = (int64_t)payload->bsize;
3714 2137 : if (nrows) *nrows = (int64_t)payload->nrows;
3715 2137 : return payload->buffer;
3716 : }
3717 :
3718 521595 : static int cloudsync_payload_decode_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
3719 521595 : cloudsync_pk_decode_bind_context *decode_context = (cloudsync_pk_decode_bind_context*)xdata;
3720 521595 : int rc = pk_decode_bind_callback(decode_context->vm, index, type, ival, dval, pval);
3721 :
3722 521595 : if (rc == DBRES_OK) {
3723 : // the dbversion index is smaller than seq index, so it is processed first
3724 : // when processing the dbversion column: save the value to the tmp_dbversion field
3725 : // when processing the seq column: update the dbversion and seq fields only if the current dbversion is greater than the last max value
3726 521595 : switch (index) {
3727 : case CLOUDSYNC_PK_INDEX_TBL:
3728 57955 : if (type == DBTYPE_TEXT) {
3729 57955 : decode_context->tbl = pval;
3730 57955 : decode_context->tbl_len = ival;
3731 57955 : }
3732 57955 : break;
3733 : case CLOUDSYNC_PK_INDEX_PK:
3734 57955 : if (type == DBTYPE_BLOB) {
3735 57955 : decode_context->pk = pval;
3736 57955 : decode_context->pk_len = ival;
3737 57955 : }
3738 57955 : break;
3739 : case CLOUDSYNC_PK_INDEX_COLNAME:
3740 57955 : if (type == DBTYPE_TEXT) {
3741 57955 : decode_context->col_name = pval;
3742 57955 : decode_context->col_name_len = ival;
3743 57955 : }
3744 57955 : break;
3745 : case CLOUDSYNC_PK_INDEX_COLVERSION:
3746 57955 : if (type == DBTYPE_INTEGER) decode_context->col_version = ival;
3747 57955 : break;
3748 : case CLOUDSYNC_PK_INDEX_DBVERSION:
3749 57955 : if (type == DBTYPE_INTEGER) decode_context->db_version = ival;
3750 57955 : break;
3751 : case CLOUDSYNC_PK_INDEX_SITEID:
3752 57955 : if (type == DBTYPE_BLOB) {
3753 57955 : decode_context->site_id = pval;
3754 57955 : decode_context->site_id_len = ival;
3755 57955 : }
3756 57955 : break;
3757 : case CLOUDSYNC_PK_INDEX_CL:
3758 57955 : if (type == DBTYPE_INTEGER) decode_context->cl = ival;
3759 57955 : break;
3760 : case CLOUDSYNC_PK_INDEX_SEQ:
3761 57955 : if (type == DBTYPE_INTEGER) decode_context->seq = ival;
3762 57955 : break;
3763 : }
3764 521595 : }
3765 :
3766 521595 : return rc;
3767 : }
3768 :
3769 : typedef struct {
3770 : const char *tbl;
3771 : int64_t tbl_len;
3772 : const void *pk;
3773 : int64_t pk_len;
3774 : const char *col_name;
3775 : int64_t col_name_len;
3776 : const void *col_value;
3777 : int64_t col_value_len;
3778 : int64_t col_version;
3779 : int64_t db_version;
3780 : const void *site_id;
3781 : int64_t site_id_len;
3782 : int64_t cl;
3783 : int64_t seq;
3784 : } cloudsync_payload_fragment_row;
3785 :
3786 738 : static int cloudsync_payload_fragment_decode_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
3787 738 : UNUSED_PARAMETER(dval);
3788 738 : cloudsync_payload_fragment_row *row = (cloudsync_payload_fragment_row *)xdata;
3789 738 : switch (index) {
3790 : case CLOUDSYNC_PK_INDEX_TBL:
3791 82 : if (type != DBTYPE_TEXT) return DBRES_ERROR;
3792 82 : row->tbl = pval; row->tbl_len = ival;
3793 82 : break;
3794 : case CLOUDSYNC_PK_INDEX_PK:
3795 82 : if (type != DBTYPE_BLOB) return DBRES_ERROR;
3796 82 : row->pk = pval; row->pk_len = ival;
3797 82 : break;
3798 : case CLOUDSYNC_PK_INDEX_COLNAME:
3799 82 : if (type != DBTYPE_TEXT) return DBRES_ERROR;
3800 82 : row->col_name = pval; row->col_name_len = ival;
3801 82 : break;
3802 : case CLOUDSYNC_PK_INDEX_COLVALUE:
3803 82 : if (type != DBTYPE_BLOB) return DBRES_ERROR;
3804 82 : row->col_value = pval; row->col_value_len = ival;
3805 82 : break;
3806 : case CLOUDSYNC_PK_INDEX_COLVERSION:
3807 82 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3808 82 : row->col_version = ival;
3809 82 : break;
3810 : case CLOUDSYNC_PK_INDEX_DBVERSION:
3811 82 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3812 82 : row->db_version = ival;
3813 82 : break;
3814 : case CLOUDSYNC_PK_INDEX_SITEID:
3815 82 : if (type != DBTYPE_BLOB) return DBRES_ERROR;
3816 82 : row->site_id = pval; row->site_id_len = ival;
3817 82 : break;
3818 : case CLOUDSYNC_PK_INDEX_CL:
3819 82 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3820 82 : row->cl = ival;
3821 82 : break;
3822 : case CLOUDSYNC_PK_INDEX_SEQ:
3823 82 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3824 82 : row->seq = ival;
3825 82 : break;
3826 : }
3827 738 : return DBRES_OK;
3828 738 : }
3829 :
3830 164 : static bool cloudsync_payload_is_hex (const char *value, size_t len) {
3831 4100 : for (size_t i = 0; i < len; ++i) {
3832 3936 : char c = value[i];
3833 3936 : if (!((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F'))) return false;
3834 3936 : }
3835 164 : return true;
3836 164 : }
3837 :
3838 246 : static bool cloudsync_payload_parse_u64_segment (const char *start, const char *end, uint64_t max_value, uint64_t *out, const char **next) {
3839 246 : if (!start || !end || start >= end) return false;
3840 246 : uint64_t value = 0;
3841 246 : const char *p = start;
3842 909 : while (p < end && *p >= '0' && *p <= '9') {
3843 663 : uint64_t digit = (uint64_t)(*p - '0');
3844 663 : if (value > (max_value - digit) / 10u) return false;
3845 663 : value = value * 10u + digit;
3846 663 : p++;
3847 : }
3848 246 : if (p == start || p >= end || *p != ':') return false;
3849 246 : *out = value;
3850 246 : *next = p + 1;
3851 246 : return true;
3852 246 : }
3853 :
3854 82 : static bool cloudsync_payload_fragment_parse_colname (const char *col_name, int64_t col_name_len,
3855 : char *value_id, size_t value_id_len,
3856 : char *checksum_hex, size_t checksum_hex_len,
3857 : int *part_index, int *part_count,
3858 : int64_t *total_size,
3859 : const char **base_col, int64_t *base_col_len) {
3860 82 : size_t prefix_len = strlen(CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX);
3861 82 : if (!col_name || col_name_len <= (int64_t)prefix_len) return false;
3862 82 : if (strncmp(col_name, CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX, prefix_len) != 0) return false;
3863 :
3864 82 : const char *p = col_name + prefix_len;
3865 82 : const char *end = col_name + col_name_len;
3866 82 : const char *sep = memchr(p, ':', (size_t)(end - p));
3867 82 : if (!sep || (size_t)(sep - p) + 1 > value_id_len) return false;
3868 82 : if ((sep - p) != 32 || !cloudsync_payload_is_hex(p, (size_t)(sep - p))) return false;
3869 82 : memcpy(value_id, p, (size_t)(sep - p));
3870 82 : value_id[sep - p] = 0;
3871 :
3872 82 : p = sep + 1;
3873 82 : sep = memchr(p, ':', (size_t)(end - p));
3874 82 : if (!sep || (size_t)(sep - p) + 1 > checksum_hex_len) return false;
3875 82 : if ((sep - p) != 16 || !cloudsync_payload_is_hex(p, (size_t)(sep - p))) return false;
3876 82 : memcpy(checksum_hex, p, (size_t)(sep - p));
3877 82 : checksum_hex[sep - p] = 0;
3878 :
3879 82 : const char *next = NULL;
3880 82 : uint64_t parsed = 0;
3881 82 : if (!cloudsync_payload_parse_u64_segment(sep + 1, end, INT_MAX, &parsed, &next)) return false;
3882 82 : *part_index = (int)parsed;
3883 :
3884 82 : if (!cloudsync_payload_parse_u64_segment(next, end, INT_MAX, &parsed, &next)) return false;
3885 82 : *part_count = (int)parsed;
3886 :
3887 82 : if (!cloudsync_payload_parse_u64_segment(next, end, INT64_MAX, &parsed, &next)) return false;
3888 82 : *total_size = (int64_t)parsed;
3889 :
3890 82 : *base_col = next;
3891 82 : *base_col_len = end - *base_col;
3892 82 : return (*part_count > 0 && *part_index < *part_count && *base_col_len > 0);
3893 82 : }
3894 :
3895 : typedef struct {
3896 : dbvm_t *vm;
3897 : int param_index;
3898 : } cloudsync_payload_bind_param_context;
3899 :
3900 23 : static int cloudsync_payload_bind_param_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
3901 23 : UNUSED_PARAMETER(index);
3902 23 : cloudsync_payload_bind_param_context *ctx = (cloudsync_payload_bind_param_context *)xdata;
3903 23 : switch (type) {
3904 0 : case DBTYPE_INTEGER: return databasevm_bind_int(ctx->vm, ctx->param_index, ival);
3905 0 : case DBTYPE_FLOAT: return databasevm_bind_double(ctx->vm, ctx->param_index, dval);
3906 0 : case DBTYPE_NULL: return databasevm_bind_null(ctx->vm, ctx->param_index);
3907 2 : case DBTYPE_TEXT: return databasevm_bind_text(ctx->vm, ctx->param_index, pval, (int)ival);
3908 21 : case DBTYPE_BLOB: return databasevm_bind_blob(ctx->vm, ctx->param_index, pval, (uint64_t)ival);
3909 : }
3910 0 : return DBRES_MISUSE;
3911 23 : }
3912 :
3913 79 : static int cloudsync_payload_fragments_cleanup_stale (cloudsync_context *data) {
3914 : // Stale-fragment GC is pure maintenance (it removes incomplete fragment groups
3915 : // older than CLOUDSYNC_PAYLOAD_FRAGMENT_STALE_SECONDS), so it has no correctness
3916 : // deadline. It runs a full GROUP BY scan of the fragments table; calling it on
3917 : // every applied fragment would be O(n^2) for a heavily-fragmented value, since
3918 : // each fragment arrives as its own apply call. Throttle it to at most once per
3919 : // CLOUDSYNC_PAYLOAD_FRAGMENT_CLEANUP_MIN_INTERVAL per connection.
3920 : // Its own savepoint keeps a failed cleanup from failing the piece being applied:
3921 : // it is rolled back and logged. Only a savepoint that cannot be managed is an error.
3922 79 : int64_t now = (int64_t)time(NULL);
3923 79 : if (data->last_fragment_cleanup != 0 &&
3924 60 : now - data->last_fragment_cleanup < CLOUDSYNC_PAYLOAD_FRAGMENT_CLEANUP_MIN_INTERVAL) {
3925 60 : return DBRES_OK;
3926 : }
3927 19 : data->last_fragment_cleanup = now;
3928 :
3929 19 : int rc = database_begin_savepoint(data, "cloudsync_fragment_cleanup");
3930 19 : if (rc != DBRES_OK) return rc;
3931 19 : dbvm_t *vm = NULL;
3932 19 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_CLEANUP_STALE, &vm, 0);
3933 19 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 1, now - CLOUDSYNC_PAYLOAD_FRAGMENT_STALE_SECONDS);
3934 19 : if (rc == DBRES_OK) rc = databasevm_step(vm);
3935 19 : if (vm) databasevm_finalize(vm);
3936 19 : if (rc == DBRES_DONE) return database_commit_savepoint(data, "cloudsync_fragment_cleanup");
3937 :
3938 : char warning[1100];
3939 1 : snprintf(warning, sizeof(warning), "stale fragment cleanup failed: %s", cloudsync_errmsg(data));
3940 1 : rc = database_rollback_savepoint(data, "cloudsync_fragment_cleanup");
3941 1 : if (rc != DBRES_OK) return rc;
3942 1 : cloudsync_reset_error(data);
3943 1 : database_log_warning(data, warning);
3944 1 : return DBRES_OK;
3945 79 : }
3946 :
3947 23 : static int cloudsync_payload_apply_single_decoded_row (cloudsync_context *data,
3948 : const char *tbl, size_t tbl_len,
3949 : const char *pk, size_t pk_len,
3950 : const char *col_name, size_t col_name_len,
3951 : const char *encoded_value, size_t encoded_value_len,
3952 : int64_t col_version, int64_t db_version,
3953 : const char *site_id, size_t site_id_len,
3954 : int64_t cl, int64_t seq,
3955 : int *pnrows) {
3956 23 : int rc = DBRES_OK;
3957 23 : dbvm_t *vm = NULL;
3958 23 : merge_pending_batch batch = {0};
3959 :
3960 23 : rc = databasevm_prepare(data, SQL_CHANGES_INSERT_ROW, &vm, 0);
3961 23 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: error while compiling SQL statement", rc);
3962 :
3963 23 : rc = databasevm_bind_text(vm, 1, tbl, (int)tbl_len);
3964 23 : if (rc == DBRES_OK) rc = databasevm_bind_blob(vm, 2, pk, (uint64_t)pk_len);
3965 23 : if (rc == DBRES_OK) rc = databasevm_bind_text(vm, 3, col_name, (int)col_name_len);
3966 23 : if (rc == DBRES_OK) {
3967 23 : if (data->skip_decode_idx == CLOUDSYNC_PK_INDEX_COLVALUE) {
3968 0 : rc = databasevm_bind_blob(vm, 4, encoded_value, (uint64_t)encoded_value_len);
3969 0 : } else {
3970 23 : size_t seek = 0;
3971 23 : cloudsync_payload_bind_param_context bind_ctx = {.vm = vm, .param_index = 4};
3972 23 : int res = pk_decode((char *)encoded_value, encoded_value_len, 1, &seek, -1, cloudsync_payload_bind_param_callback, &bind_ctx);
3973 23 : if (res == -1 || seek != encoded_value_len) rc = cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 encoded value", DBRES_MISUSE);
3974 : }
3975 23 : }
3976 23 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 5, col_version);
3977 23 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 6, db_version);
3978 23 : if (rc == DBRES_OK) rc = databasevm_bind_blob(vm, 7, site_id, (uint64_t)site_id_len);
3979 23 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 8, cl);
3980 23 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 9, seq);
3981 23 : if (rc != DBRES_OK) goto cleanup;
3982 :
3983 23 : data->pending_batch = &batch;
3984 23 : rc = databasevm_step(vm);
3985 23 : if (rc == DBRES_DONE) rc = DBRES_OK;
3986 23 : if (rc != DBRES_OK) {
3987 0 : cloudsync_set_dberror(data);
3988 0 : goto cleanup;
3989 : }
3990 :
3991 23 : rc = merge_flush_pending(data);
3992 23 : if (rc != DBRES_OK) goto cleanup;
3993 22 : data->pending_batch = NULL;
3994 :
3995 : // Do NOT advance the receive cursor here: a v3 value carries a single
3996 : // (db_version, seq) that can be in the middle of its source db_version, and a
3997 : // db_version's chunks can span multiple /check artifacts. Advancing per value
3998 : // would leave the cursor mid-db_version. The durable cursor is advanced once,
3999 : // after the whole payload/stream is applied, via cloudsync_payload_apply's
4000 : // checkpoint argument. Record the last applied position for the
4001 : // CLOUDSYNC_CHECKPOINT_LAST_APPLIED mode.
4002 22 : if (db_version > data->apply_last_db_version ||
4003 0 : (db_version == data->apply_last_db_version && seq > data->apply_last_seq)) {
4004 22 : data->apply_last_db_version = db_version;
4005 22 : data->apply_last_seq = seq;
4006 22 : }
4007 :
4008 22 : if (pnrows) *pnrows += 1;
4009 :
4010 : cleanup:
4011 23 : data->pending_batch = NULL;
4012 23 : merge_pending_free_entries(&batch);
4013 23 : if (batch.cached_vm) databasevm_finalize(batch.cached_vm);
4014 23 : if (batch.cached_col_names) cloudsync_memory_free(batch.cached_col_names);
4015 23 : if (batch.entries) cloudsync_memory_free(batch.entries);
4016 23 : if (vm) databasevm_finalize(vm);
4017 23 : return rc;
4018 23 : }
4019 :
4020 79 : static int cloudsync_payload_apply_reassembled_fragment (cloudsync_context *data, const char *value_id, const char *expected_checksum_hex, int *pnrows) {
4021 79 : int rc = DBRES_OK;
4022 79 : dbvm_t *vm = NULL;
4023 79 : char *value = NULL;
4024 79 : char *tbl = NULL, *col_name = NULL;
4025 79 : char *pk = NULL, *site_id = NULL;
4026 79 : size_t tbl_len = 0, col_name_len = 0, pk_len = 0, site_id_len = 0;
4027 79 : int64_t col_version = 0, db_version = 0, cl = 0, seq = 0;
4028 79 : int64_t total_size = 0, copied = 0;
4029 :
4030 79 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_COUNT, &vm, 0);
4031 79 : if (rc != DBRES_OK) return rc;
4032 79 : rc = databasevm_bind_text(vm, 1, value_id, -1);
4033 79 : if (rc != DBRES_OK) { databasevm_finalize(vm); return rc; }
4034 79 : rc = databasevm_step(vm);
4035 79 : if (rc != DBRES_ROW) {
4036 0 : cloudsync_set_dberror(data);
4037 0 : databasevm_finalize(vm);
4038 0 : return (rc == DBRES_DONE) ? DBRES_OK : rc;
4039 : }
4040 79 : int64_t have = database_column_int(vm, 0);
4041 79 : int64_t part_count_min = database_column_int(vm, 1);
4042 79 : int64_t part_count_max = database_column_int(vm, 2);
4043 79 : int64_t total_size_min = database_column_int(vm, 3);
4044 79 : int64_t total_size_max = database_column_int(vm, 4);
4045 79 : const char *checksum_min = database_column_text(vm, 5);
4046 79 : const char *checksum_max = database_column_text(vm, 6);
4047 79 : char checksum_min_copy[32] = {0};
4048 79 : char checksum_max_copy[32] = {0};
4049 79 : if (checksum_min) snprintf(checksum_min_copy, sizeof(checksum_min_copy), "%s", checksum_min);
4050 79 : if (checksum_max) snprintf(checksum_max_copy, sizeof(checksum_max_copy), "%s", checksum_max);
4051 79 : int64_t part_index_min = database_column_int(vm, 7);
4052 79 : int64_t part_index_max = database_column_int(vm, 8);
4053 79 : databasevm_finalize(vm);
4054 79 : vm = NULL;
4055 79 : if (have <= 0 || part_count_min <= 0 || have < part_count_max) return DBRES_OK;
4056 46 : if (part_count_min != part_count_max || total_size_min != total_size_max || !checksum_min_copy[0] || !checksum_max_copy[0] ||
4057 23 : strcmp(checksum_min_copy, checksum_max_copy) != 0 || strcmp(checksum_min_copy, expected_checksum_hex) != 0 ||
4058 23 : part_index_min != 0 || part_index_max != part_count_max - 1 || have != part_count_max) {
4059 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: inconsistent v3 fragments", DBRES_MISUSE);
4060 : }
4061 23 : total_size = total_size_max;
4062 :
4063 23 : value = cloudsync_memory_alloc((uint64_t)total_size);
4064 23 : if (!value) return DBRES_NOMEM;
4065 :
4066 23 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_SELECT, &vm, 0);
4067 23 : if (rc != DBRES_OK) goto cleanup;
4068 23 : rc = databasevm_bind_text(vm, 1, value_id, -1);
4069 23 : if (rc != DBRES_OK) goto cleanup;
4070 :
4071 23 : uint64_t checksum = 14695981039346656037ULL;
4072 96 : while ((rc = databasevm_step(vm)) == DBRES_ROW) {
4073 73 : size_t frag_len = 0;
4074 73 : const char *frag = database_column_blob(vm, 0, &frag_len);
4075 73 : if (copied + (int64_t)frag_len > total_size) { rc = DBRES_MISUSE; goto cleanup; }
4076 73 : memcpy(value + copied, frag, frag_len);
4077 73 : checksum = cloudsync_checksum_update(checksum, frag, frag_len);
4078 73 : copied += (int64_t)frag_len;
4079 :
4080 73 : if (!tbl) {
4081 23 : const char *t = database_column_text(vm, 1);
4082 23 : const char *c = database_column_text(vm, 3);
4083 23 : size_t pkl = 0, sidl = 0;
4084 23 : const char *p = database_column_blob(vm, 2, &pkl);
4085 23 : const char *sid = database_column_blob(vm, 6, &sidl);
4086 23 : tbl_len = (size_t)database_column_bytes(vm, 1);
4087 23 : col_name_len = (size_t)database_column_bytes(vm, 3);
4088 23 : pk_len = pkl;
4089 23 : site_id_len = sidl;
4090 23 : tbl = cloudsync_memory_alloc((uint64_t)tbl_len);
4091 23 : col_name = cloudsync_memory_alloc((uint64_t)col_name_len);
4092 23 : pk = cloudsync_memory_alloc((uint64_t)pk_len);
4093 23 : site_id = cloudsync_memory_alloc((uint64_t)site_id_len);
4094 23 : if (!tbl || !col_name || !pk || !site_id) { rc = DBRES_NOMEM; goto cleanup; }
4095 23 : memcpy(tbl, t, tbl_len);
4096 23 : memcpy(col_name, c, col_name_len);
4097 23 : memcpy(pk, p, pk_len);
4098 23 : memcpy(site_id, sid, site_id_len);
4099 23 : col_version = database_column_int(vm, 4);
4100 23 : db_version = database_column_int(vm, 5);
4101 23 : cl = database_column_int(vm, 7);
4102 23 : seq = database_column_int(vm, 8);
4103 23 : } else {
4104 50 : size_t pkl = 0, sidl = 0;
4105 50 : const char *t = database_column_text(vm, 1);
4106 50 : const char *c = database_column_text(vm, 3);
4107 50 : const char *p = database_column_blob(vm, 2, &pkl);
4108 50 : const char *sid = database_column_blob(vm, 6, &sidl);
4109 100 : if ((size_t)database_column_bytes(vm, 1) != tbl_len || memcmp(tbl, t, tbl_len) != 0 ||
4110 50 : pkl != pk_len || memcmp(pk, p, pk_len) != 0 ||
4111 50 : (size_t)database_column_bytes(vm, 3) != col_name_len || memcmp(col_name, c, col_name_len) != 0 ||
4112 50 : database_column_int(vm, 4) != col_version ||
4113 50 : database_column_int(vm, 5) != db_version ||
4114 50 : sidl != site_id_len || memcmp(site_id, sid, site_id_len) != 0 ||
4115 50 : database_column_int(vm, 7) != cl ||
4116 50 : database_column_int(vm, 8) != seq) {
4117 0 : rc = DBRES_MISUSE;
4118 0 : goto cleanup;
4119 : }
4120 : }
4121 : }
4122 23 : if (rc == DBRES_DONE) rc = DBRES_OK;
4123 23 : if (rc != DBRES_OK) goto cleanup;
4124 23 : if (copied != total_size) { rc = DBRES_MISUSE; goto cleanup; }
4125 : char checksum_hex[17];
4126 23 : snprintf(checksum_hex, sizeof(checksum_hex), "%016" PRIx64, checksum);
4127 23 : if (strcmp(checksum_hex, expected_checksum_hex) != 0) { rc = DBRES_MISUSE; goto cleanup; }
4128 23 : databasevm_finalize(vm);
4129 23 : vm = NULL;
4130 :
4131 46 : rc = cloudsync_payload_apply_single_decoded_row(data, tbl, tbl_len, pk, pk_len, col_name, col_name_len,
4132 23 : value, (size_t)total_size, col_version, db_version,
4133 23 : site_id, site_id_len, cl, seq, pnrows);
4134 23 : if (rc != DBRES_OK) goto cleanup;
4135 :
4136 : // the caller's savepoint makes applying the value and removing its pieces one unit
4137 22 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_DELETE, &vm, 0);
4138 22 : if (rc == DBRES_OK) rc = databasevm_bind_text(vm, 1, value_id, -1);
4139 22 : if (rc == DBRES_OK) rc = databasevm_step(vm);
4140 23 : if (rc == DBRES_DONE) rc = DBRES_OK;
4141 1 : else cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to remove applied fragments", rc);
4142 :
4143 : cleanup:
4144 23 : if (vm) databasevm_finalize(vm);
4145 23 : if (value) cloudsync_memory_free(value);
4146 23 : if (tbl) cloudsync_memory_free(tbl);
4147 23 : if (col_name) cloudsync_memory_free(col_name);
4148 23 : if (pk) cloudsync_memory_free(pk);
4149 23 : if (site_id) cloudsync_memory_free(site_id);
4150 23 : return rc;
4151 79 : }
4152 :
4153 82 : static int cloudsync_payload_apply_fragment_row (cloudsync_context *data, cloudsync_payload_fragment_row *row, bool track, int *pnrows) {
4154 : char value_id[64];
4155 : char checksum_hex[17];
4156 82 : int part_index = 0, part_count = 0;
4157 82 : int64_t total_size = 0;
4158 82 : const char *base_col = NULL;
4159 82 : int64_t base_col_len = 0;
4160 164 : if (!row || !row->tbl || row->tbl_len <= 0 || !row->pk || row->pk_len <= 0 ||
4161 82 : !row->col_name || row->col_name_len <= 0 || !row->col_value || row->col_value_len <= 0 ||
4162 82 : !row->site_id || row->site_id_len <= 0) {
4163 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 payload row", DBRES_MISUSE);
4164 : }
4165 164 : if (!cloudsync_payload_fragment_parse_colname(row->col_name, row->col_name_len, value_id, sizeof(value_id),
4166 82 : checksum_hex, sizeof(checksum_hex),
4167 : &part_index, &part_count, &total_size, &base_col, &base_col_len)) {
4168 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 fragment metadata", DBRES_MISUSE);
4169 : }
4170 :
4171 82 : uint64_t value_checksum = strtoull(checksum_hex, NULL, 16);
4172 : char expected_value_id[33];
4173 164 : cloudsync_payload_fragment_value_id(expected_value_id, row->tbl, (int)row->tbl_len, row->pk, (int)row->pk_len,
4174 82 : base_col, (int)base_col_len, row->col_version, row->db_version,
4175 82 : row->site_id, (int)row->site_id_len, row->cl, row->seq,
4176 82 : value_checksum, total_size);
4177 82 : if (strcmp(value_id, expected_value_id) != 0) {
4178 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 fragment identity", DBRES_MISUSE);
4179 : }
4180 :
4181 : // A piece of a value this stream already applied is redundant: staging it again
4182 : // would leave it behind as an incomplete group. It happens when pieces left by an
4183 : // interrupted attempt complete the value before the replay delivers its last piece.
4184 82 : cloudsync_stream_value *seen = track ? cloudsync_stream_value_find(data, value_id) : NULL;
4185 82 : if (seen && seen->applied) return DBRES_OK;
4186 :
4187 : // concurrent transactions applying pieces of this value wait for each other
4188 79 : int rc = database_fragment_lock(data, value_id);
4189 79 : if (rc != DBRES_OK) return rc;
4190 :
4191 : // Stage the piece, then reassemble, apply and remove the value's pieces as one
4192 : // unit: a failure rolls back only this call and leaves the pieces staged by
4193 : // earlier calls, so the value stays retryable.
4194 79 : rc = database_begin_savepoint(data, "cloudsync_fragment");
4195 79 : if (rc != DBRES_OK) return rc;
4196 :
4197 : // the fragments table is guaranteed by dbutils_settings_init; no DDL here
4198 : // because the apply path runs under sync-only credentials on server nodes
4199 79 : dbvm_t *vm = NULL;
4200 79 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_UPSERT, &vm, 0);
4201 79 : if (rc == DBRES_OK) {
4202 79 : databasevm_bind_text(vm, 1, value_id, -1);
4203 79 : databasevm_bind_int(vm, 2, part_index);
4204 79 : databasevm_bind_int(vm, 3, part_count);
4205 79 : databasevm_bind_int(vm, 4, total_size);
4206 79 : databasevm_bind_text(vm, 5, checksum_hex, -1);
4207 79 : databasevm_bind_int(vm, 6, (int64_t)time(NULL));
4208 79 : databasevm_bind_text(vm, 7, row->tbl, (int)row->tbl_len);
4209 79 : databasevm_bind_blob(vm, 8, row->pk, (uint64_t)row->pk_len);
4210 79 : databasevm_bind_text(vm, 9, base_col, (int)base_col_len);
4211 79 : databasevm_bind_int(vm, 10, row->col_version);
4212 79 : databasevm_bind_int(vm, 11, row->db_version);
4213 79 : databasevm_bind_blob(vm, 12, row->site_id, (uint64_t)row->site_id_len);
4214 79 : databasevm_bind_int(vm, 13, row->cl);
4215 79 : databasevm_bind_int(vm, 14, row->seq);
4216 79 : databasevm_bind_blob(vm, 15, row->col_value, (uint64_t)row->col_value_len);
4217 79 : rc = databasevm_step(vm);
4218 79 : if (rc == DBRES_DONE) rc = DBRES_OK;
4219 0 : else cloudsync_set_dberror(data);
4220 79 : }
4221 79 : if (vm) databasevm_finalize(vm);
4222 :
4223 : // After staging, so a group resumed after a long pause is recent again and is kept.
4224 79 : if (rc == DBRES_OK) rc = cloudsync_payload_fragments_cleanup_stale(data);
4225 :
4226 79 : int applied = 0;
4227 79 : if (rc == DBRES_OK) rc = cloudsync_payload_apply_reassembled_fragment(data, value_id, checksum_hex, &applied);
4228 79 : if (rc == DBRES_OK) rc = database_commit_savepoint(data, "cloudsync_fragment");
4229 79 : if (rc != DBRES_OK) {
4230 : char message[1024];
4231 2 : snprintf(message, sizeof(message), "%s", cloudsync_errmsg(data));
4232 2 : int sqlstate = cloudsync_sqlstate(data);
4233 2 : database_rollback_savepoint(data, "cloudsync_fragment");
4234 2 : cloudsync_reset_error(data);
4235 2 : cloudsync_set_error(data, message[0] ? message : "Unable to apply a fragment", rc);
4236 2 : cloudsync_set_sqlstate(data, sqlstate);
4237 2 : return rc;
4238 : }
4239 77 : if (pnrows) *pnrows += applied;
4240 77 : return track ? cloudsync_stream_value_track(data, value_id, applied > 0) : DBRES_OK;
4241 82 : }
4242 :
4243 : // #ifndef CLOUDSYNC_OMIT_RLS_VALIDATION
4244 :
4245 : // Called when a receive stream's final chunk has applied. A fragmented value the stream
4246 : // delivered must be complete by now: its pieces are removed when it is applied, here or
4247 : // by another connection. Only values this stream staged are checked. Staging left by
4248 : // other streams or direct calls is ignored on purpose: a value_id identifies a value,
4249 : // not the stream that delivered it, and a replay can legitimately omit an old value
4250 : // (replaced, lost a conflict, filtered out, untracked), so requiring the whole staging
4251 : // table to be empty could stop the cursor for good. The age cleanup bounds such
4252 : // leftovers. A value another connection completes while this stream is still
4253 : // delivering it leaves this stream's later pieces staged again: the check fails once,
4254 : // and the replay from the first page applies the value again as a no-op and passes.
4255 13 : static int cloudsync_receive_stream_check_complete (cloudsync_context *data) {
4256 17 : for (int i = 0; i < data->stream_values_count; i++) {
4257 6 : if (data->stream_values[i].applied) continue;
4258 2 : dbvm_t *vm = NULL;
4259 2 : int rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_EXISTS, &vm, 0);
4260 2 : if (rc == DBRES_OK) rc = databasevm_bind_text(vm, 1, data->stream_values[i].value_id, -1);
4261 2 : if (rc == DBRES_OK) rc = databasevm_step(vm);
4262 2 : if (vm) databasevm_finalize(vm);
4263 2 : if (rc == DBRES_ROW) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: the receive stream ended with an incomplete fragmented value", DBRES_ERROR);
4264 0 : if (rc != DBRES_DONE) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to check staged fragments", rc);
4265 0 : }
4266 11 : return DBRES_OK;
4267 13 : }
4268 :
4269 : // Advance the durable receive cursor (check_dbversion/check_seq) after a payload
4270 : // (or a fully-applied chunk stream) has been applied. See the checkpoint-mode
4271 : // documentation on cloudsync_payload_apply in cloudsync.h. The advance is
4272 : // strictly monotonic so re-delivered rows never regress the cursor.
4273 1589 : static int cloudsync_payload_apply_checkpoint (cloudsync_context *data, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
4274 : int64_t target_db_version;
4275 : int64_t target_seq;
4276 :
4277 1589 : if (checkpoint_db_version == CLOUDSYNC_CHECKPOINT_NONE) return DBRES_OK;
4278 1563 : bool stream_end = (checkpoint_db_version >= 0 || checkpoint_db_version == CLOUDSYNC_CHECKPOINT_STREAM_LEGACY);
4279 1563 : if (stream_end) {
4280 13 : int rc = cloudsync_receive_stream_check_complete(data);
4281 13 : if (rc != DBRES_OK) return rc;
4282 11 : }
4283 1561 : if (checkpoint_db_version < 0) {
4284 : // Nothing applied -> nothing to checkpoint.
4285 1551 : if (data->apply_last_db_version < 0) {
4286 0 : if (stream_end) cloudsync_receive_stream_reset(data);
4287 0 : return DBRES_OK;
4288 : }
4289 1551 : target_db_version = data->apply_last_db_version;
4290 1551 : target_seq = data->apply_last_seq;
4291 1551 : } else {
4292 10 : target_db_version = checkpoint_db_version;
4293 10 : target_seq = checkpoint_seq;
4294 : }
4295 :
4296 1561 : int64_t cur_db_version = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION);
4297 1561 : int64_t cur_seq = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_CHECK_SEQ);
4298 :
4299 : // monotonic guard: never move the cursor backwards
4300 1561 : if (target_db_version > cur_db_version || (target_db_version == cur_db_version && target_seq > cur_seq)) {
4301 : // db_version first: failing between the two writes then re-delivers rows
4302 : // instead of skipping them
4303 : char buf[256];
4304 1101 : snprintf(buf, sizeof(buf), "%" PRId64, target_db_version);
4305 1101 : int rc = dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION, buf);
4306 1101 : if (rc == DBRES_OK && target_seq != cur_seq) {
4307 813 : snprintf(buf, sizeof(buf), "%" PRId64, target_seq);
4308 813 : rc = dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_SEQ, buf);
4309 813 : }
4310 1101 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to write the receive checkpoint", rc);
4311 1100 : }
4312 1560 : if (stream_end) cloudsync_receive_stream_reset(data);
4313 1560 : return DBRES_OK;
4314 1589 : }
4315 :
4316 : // Steps one decoded payload row (an INSERT into cloudsync_changes). On PostgreSQL the merge
4317 : // runs inside the cloudsync_changes trigger, so the row gets its own savepoint: a failed
4318 : // write rolls back cleanly, metadata included, and leaves the transaction usable for the
4319 : // PK group's rollback. The failure's message and SQLSTATE survive the rollback.
4320 57902 : static int cloudsync_payload_apply_row (cloudsync_context *data, dbvm_t *vm) {
4321 : #ifdef CLOUDSYNC_POSTGRESQL_BUILD
4322 : int rc = database_begin_savepoint(data, "cloudsync_apply_row");
4323 : if (rc != DBRES_OK) return rc;
4324 : rc = databasevm_step(vm);
4325 : if (rc == DBRES_DONE) {
4326 : int commit_rc = database_commit_savepoint(data, "cloudsync_apply_row");
4327 : if (commit_rc == DBRES_OK) return DBRES_DONE;
4328 : rc = commit_rc;
4329 : }
4330 : char message[1024];
4331 : snprintf(message, sizeof(message), "%s", cloudsync_errmsg(data));
4332 : int sqlstate = cloudsync_sqlstate(data);
4333 : database_rollback_savepoint(data, "cloudsync_apply_row");
4334 : cloudsync_reset_error(data);
4335 : cloudsync_set_error(data, message[0] ? message : "Unable to apply a received change", rc);
4336 : cloudsync_set_sqlstate(data, sqlstate);
4337 : return rc;
4338 : #else
4339 57902 : return databasevm_step(vm);
4340 : #endif
4341 : }
4342 :
4343 : // True when the decoded payload row writes one block of a block column. Its merge
4344 : // materializes the column into the base table, so the row's other pending columns must
4345 : // be written first: materializing into a row that does not exist yet has to insert it
4346 : // with nothing but the primary key and that column.
4347 14528 : static bool cloudsync_payload_row_is_block (const cloudsync_pk_decode_bind_context *row) {
4348 29056 : return row->col_name && row->col_name_len > 0 &&
4349 14528 : memchr(row->col_name, BLOCK_SEPARATOR, (size_t)row->col_name_len) != NULL;
4350 : }
4351 :
4352 : // Do not write group metadata unless its rollback boundary was established.
4353 57902 : static int cloudsync_payload_group_open (cloudsync_context *data, merge_pending_batch *batch) {
4354 57902 : if (batch->group_savepoint) return DBRES_OK;
4355 21958 : int rc = database_begin_savepoint(data, "cloudsync_merge_group");
4356 21958 : batch->group_savepoint = (rc == DBRES_OK);
4357 21958 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to start a payload group", rc);
4358 21958 : return rc;
4359 57902 : }
4360 :
4361 : // Flushes the pending PK group and closes its savepoint: released when the flush
4362 : // succeeds, rolled back when it fails, so a failed group leaves no trace behind and
4363 : // applies cleanly when delivered again. The failure's message and SQLSTATE survive the
4364 : // rollback.
4365 23668 : static int cloudsync_payload_group_flush (cloudsync_context *data, merge_pending_batch *batch) {
4366 23668 : int rc = merge_flush_pending(data);
4367 23668 : if (!batch->group_savepoint) return rc;
4368 21836 : batch->group_savepoint = false;
4369 21836 : if (rc == DBRES_OK) {
4370 21824 : rc = database_commit_savepoint(data, "cloudsync_merge_group");
4371 21824 : if (rc == DBRES_OK) return DBRES_OK;
4372 0 : }
4373 : char message[1024];
4374 12 : snprintf(message, sizeof(message), "%s", cloudsync_errmsg(data));
4375 12 : int sqlstate = cloudsync_sqlstate(data);
4376 12 : database_rollback_savepoint(data, "cloudsync_merge_group");
4377 12 : cloudsync_reset_error(data);
4378 12 : cloudsync_set_error(data, message[0] ? message : "Unable to flush pending changes", rc);
4379 12 : cloudsync_set_sqlstate(data, sqlstate);
4380 12 : return rc;
4381 23668 : }
4382 :
4383 : // Abandons an open PK group after the apply stopped: nothing it wrote is kept.
4384 390 : static void cloudsync_payload_group_abandon (cloudsync_context *data, merge_pending_batch *batch) {
4385 390 : merge_pending_free_entries(batch);
4386 390 : if (!batch->group_savepoint) return;
4387 122 : batch->group_savepoint = false;
4388 : char message[1024];
4389 122 : snprintf(message, sizeof(message), "%s", cloudsync_errmsg(data));
4390 122 : int sqlstate = cloudsync_sqlstate(data);
4391 122 : int code = cloudsync_errcode(data);
4392 122 : database_rollback_savepoint(data, "cloudsync_merge_group");
4393 122 : if (code != DBRES_OK) {
4394 120 : cloudsync_reset_error(data);
4395 120 : cloudsync_set_error(data, message, code);
4396 120 : cloudsync_set_sqlstate(data, sqlstate);
4397 120 : }
4398 390 : }
4399 :
4400 1923 : int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *pnrows, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
4401 : // Guard against calling payload_apply before cloudsync_init: without this,
4402 : // the settings lookups at the top of this function would each emit a
4403 : // "no such table: cloudsync_settings" debug line, control would fall
4404 : // through to the meta-table insert, and the function would ultimately
4405 : // return an error with an empty errmsg — SQLite then surfaces that as
4406 : // the confusing "Runtime error: not an error".
4407 1923 : if (!cloudsync_context_is_initialized(data)) {
4408 1 : return cloudsync_set_error(data,
4409 : "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
4410 : "to enable sync on a table before calling cloudsync_payload_apply().",
4411 : DBRES_MISUSE);
4412 : }
4413 :
4414 : // sanity check
4415 1922 : if (blen < (int)sizeof(cloudsync_payload_header)) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid payload length", DBRES_MISUSE);
4416 :
4417 : // track the last (db_version, seq) applied by this call so the receive
4418 : // checkpoint can be computed once, after the whole payload is applied
4419 1922 : data->apply_last_db_version = -1;
4420 1922 : data->apply_last_seq = -1;
4421 :
4422 : // decode header
4423 : cloudsync_payload_header header;
4424 1922 : memcpy(&header, payload, sizeof(cloudsync_payload_header));
4425 :
4426 1922 : header.signature = ntohl(header.signature);
4427 1922 : header.expanded_size = ntohl(header.expanded_size);
4428 1922 : header.ncols = ntohs(header.ncols);
4429 1922 : header.nrows = ntohl(header.nrows);
4430 1922 : header.schema_hash = ntohll(header.schema_hash);
4431 :
4432 : // compare schema_hash only if not disabled and if the received payload was created with the current header version
4433 : // to avoid schema hash mismatch when processed by a peer with a different extension version during software updates.
4434 1922 : if (dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_SKIP_SCHEMA_HASH_CHECK) == 0 && header.version == CLOUDSYNC_PAYLOAD_VERSION_LATEST ) {
4435 1835 : if (header.schema_hash != data->schema_hash) {
4436 4 : if (!database_check_schema_hash(data, header.schema_hash)) {
4437 : char buffer[1024];
4438 2 : snprintf(buffer, sizeof(buffer), "Cannot apply the received payload because the schema hash is unknown %" PRIu64 ".", header.schema_hash);
4439 2 : return cloudsync_set_error(data, buffer, DBRES_MISUSE);
4440 : }
4441 2 : }
4442 1833 : }
4443 :
4444 : // sanity check header
4445 1920 : if ((header.signature != CLOUDSYNC_PAYLOAD_SIGNATURE) || (header.ncols == 0)) {
4446 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid signature or column size", DBRES_MISUSE);
4447 : }
4448 1920 : if (header.version < CLOUDSYNC_PAYLOAD_VERSION_1 || header.version > CLOUDSYNC_PAYLOAD_VERSION_3) {
4449 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unsupported payload version", DBRES_MISUSE);
4450 : }
4451 :
4452 1920 : const char *buffer = payload + sizeof(cloudsync_payload_header);
4453 1920 : size_t buf_len = (size_t)blen - sizeof(cloudsync_payload_header);
4454 :
4455 : // sanity check checksum (only if version is >= 2)
4456 1920 : if (header.version >= CLOUDSYNC_PAYLOAD_MIN_VERSION_WITH_CHECKSUM) {
4457 1916 : uint64_t checksum = pk_checksum(buffer, buf_len);
4458 1916 : if (cloudsync_payload_checksum_verify(&header, checksum) == false) {
4459 1 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid checksum", DBRES_MISUSE);
4460 : }
4461 1915 : }
4462 :
4463 : // check if payload is compressed
4464 1919 : char *clone = NULL;
4465 1919 : if (header.expanded_size != 0) {
4466 : // The declared size is untrusted and is allocated before decompressing, so bound
4467 : // it by what the compressed bytes can actually expand to: a few forged header bytes
4468 : // must not make us allocate gigabytes. A genuinely large payload stays loadable,
4469 : // whatever its size. INT_MAX is LZ4's own API limit: past it the cast below turns
4470 : // negative and LZ4_decompress_safe does not validate a negative capacity.
4471 1724 : if (header.expanded_size > INT_MAX ||
4472 1723 : (uint64_t)header.expanded_size > (uint64_t)buf_len * CLOUDSYNC_PAYLOAD_LZ4_MAX_RATIO + 64) {
4473 3 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: declared expanded size is inconsistent with the compressed payload", DBRES_MISUSE);
4474 : }
4475 1721 : clone = (char *)cloudsync_memory_alloc(header.expanded_size);
4476 1721 : if (!clone) return cloudsync_set_error(data, "Unable to allocate memory to uncompress payload", DBRES_NOMEM);
4477 :
4478 1721 : int lz4_rc = LZ4_decompress_safe(buffer, clone, (int)buf_len, (int)header.expanded_size);
4479 1721 : if (lz4_rc <= 0 || (uint32_t)lz4_rc != header.expanded_size) {
4480 1 : if (clone) cloudsync_memory_free(clone);
4481 1 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to decompress BLOB", DBRES_MISUSE);
4482 : }
4483 :
4484 1720 : buffer = (const char *)clone;
4485 1720 : buf_len = (size_t)header.expanded_size;
4486 1720 : }
4487 :
4488 1915 : if (header.version == CLOUDSYNC_PAYLOAD_VERSION_3) {
4489 83 : int rc = DBRES_OK;
4490 83 : int applied_rows = 0;
4491 83 : if (header.ncols != CLOUDSYNC_CHANGES_NCOLS) {
4492 0 : if (clone) cloudsync_memory_free(clone);
4493 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 column count", DBRES_MISUSE);
4494 : }
4495 : // Without a watermark the stream end falls back to the last applied position,
4496 : // but a fragment's final chunk may apply nothing new (its pieces belong to a
4497 : // value this stream already applied), which leaves no position to checkpoint:
4498 : // fail instead of leaving the cursor in place and replaying the same window.
4499 83 : if (checkpoint_db_version == CLOUDSYNC_CHECKPOINT_STREAM_LEGACY) {
4500 1 : if (clone) cloudsync_memory_free(clone);
4501 1 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: the final chunk of a fragmented stream has no watermark", DBRES_MISUSE);
4502 : }
4503 162 : for (uint32_t i = 0; i < header.nrows; ++i) {
4504 82 : size_t seek = 0;
4505 82 : cloudsync_payload_fragment_row row = {0};
4506 82 : int res = pk_decode((char *)buffer, buf_len, header.ncols, &seek, -1,
4507 : cloudsync_payload_fragment_decode_callback, &row);
4508 82 : if (res == -1 || seek == 0 || seek > buf_len) {
4509 0 : rc = cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 payload row", DBRES_MISUSE);
4510 0 : break;
4511 : }
4512 82 : int n = 0;
4513 : // each value takes its own db_version, as each group does in the row path
4514 82 : cloudsync_pending_version_close(data);
4515 82 : rc = cloudsync_payload_apply_fragment_row(data, &row, checkpoint_db_version != CLOUDSYNC_CHECKPOINT_LAST_APPLIED, &n);
4516 : // stop at the first error, as the row path does
4517 82 : if (rc != DBRES_OK) break;
4518 80 : applied_rows += n;
4519 80 : buffer += seek;
4520 80 : buf_len -= seek;
4521 80 : }
4522 82 : cloudsync_pending_version_close(data);
4523 82 : if (clone) cloudsync_memory_free(clone);
4524 82 : cloudsync_apply_stats_add(data, applied_rows);
4525 82 : if (pnrows) *pnrows = applied_rows;
4526 : // Advance the receive cursor only after the whole payload is applied,
4527 : // gated on the caller-supplied checkpoint (a non-final chunk passes
4528 : // CLOUDSYNC_CHECKPOINT_NONE and leaves the cursor untouched). A fragment
4529 : // carries no end-of-stream marker, so LAST_APPLIED (a direct call) never
4530 : // moves the cursor.
4531 82 : if (rc == DBRES_OK && checkpoint_db_version != CLOUDSYNC_CHECKPOINT_LAST_APPLIED) {
4532 21 : rc = cloudsync_payload_apply_checkpoint(data, checkpoint_db_version, checkpoint_seq);
4533 21 : }
4534 82 : return rc;
4535 : }
4536 :
4537 : // precompile the insert statement
4538 1832 : dbvm_t *vm = NULL;
4539 1832 : int rc = databasevm_prepare(data, SQL_CHANGES_INSERT_ROW, &vm, 0);
4540 1832 : if (rc != DBRES_OK) {
4541 0 : if (clone) cloudsync_memory_free(clone);
4542 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: error while compiling SQL statement", rc);
4543 : }
4544 :
4545 : // process buffer, one row at a time
4546 1832 : uint16_t ncols = header.ncols;
4547 1832 : uint32_t nrows = header.nrows;
4548 1832 : int64_t last_payload_db_version = -1;
4549 : // The apply stops at the first failed write. Rows before the current PK group are
4550 : // written; the failed group is rolled back. applied counts the written rows.
4551 1832 : int fail_rc = DBRES_OK;
4552 1832 : int applied = 0;
4553 1832 : int applied_at_savepoint = 0; // applied when the per-db_version savepoint began
4554 1832 : cloudsync_pk_decode_bind_context decoded_context = {.vm = vm};
4555 :
4556 : // Initialize deferred column-batch merge
4557 1832 : merge_pending_batch batch = {0};
4558 1832 : data->pending_batch = &batch;
4559 1832 : bool in_savepoint = false;
4560 1832 : const void *last_pk = NULL;
4561 1832 : int64_t last_pk_len = 0;
4562 1832 : const char *last_tbl = NULL;
4563 1832 : int64_t last_tbl_len = 0;
4564 :
4565 59612 : for (uint32_t i=0; i<nrows; ++i) {
4566 57955 : size_t seek = 0;
4567 57955 : int res = pk_decode((char *)buffer, buf_len, ncols, &seek, data->skip_decode_idx, cloudsync_payload_decode_callback, &decoded_context);
4568 57955 : if (res == -1) {
4569 0 : rc = cloudsync_set_error(data, "Unable to decode a payload row", DBRES_ERROR);
4570 0 : goto cleanup;
4571 : }
4572 :
4573 : // Detect PK/table/db_version boundary to flush pending batch
4574 114078 : bool pk_changed = (last_pk != NULL &&
4575 56123 : (last_pk_len != decoded_context.pk_len ||
4576 53844 : memcmp(last_pk, decoded_context.pk, last_pk_len) != 0));
4577 114078 : bool tbl_changed = (last_tbl != NULL &&
4578 56123 : (last_tbl_len != decoded_context.tbl_len ||
4579 56051 : memcmp(last_tbl, decoded_context.tbl, last_tbl_len) != 0));
4580 57955 : bool db_version_changed = (last_payload_db_version != decoded_context.db_version);
4581 :
4582 : // Flush pending batch before any boundary change, closing the PK group
4583 57955 : if (pk_changed || tbl_changed || db_version_changed) {
4584 22011 : fail_rc = cloudsync_payload_group_flush(data, &batch);
4585 22011 : if (fail_rc != DBRES_OK) break;
4586 22008 : applied = (int)i;
4587 22008 : }
4588 :
4589 : // Per-db_version savepoints group rows with the same source db_version
4590 : // into one transaction. In SQLite autocommit mode, the RELEASE triggers
4591 : // the commit hook which bumps data->db_version and resets seq, ensuring
4592 : // unique (db_version, seq) tuples across groups. In PostgreSQL SPI,
4593 : // database_in_transaction() is always true so this block is inactive —
4594 : // the per-PK group savepoint protects each group instead.
4595 57952 : if (in_savepoint && db_version_changed) {
4596 5221 : rc = database_commit_savepoint(data, "cloudsync_payload_apply");
4597 5221 : if (rc != DBRES_OK) {
4598 50 : cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to release a savepoint", rc);
4599 50 : goto cleanup;
4600 : }
4601 5171 : in_savepoint = false;
4602 5171 : }
4603 :
4604 : // Inside a caller's transaction (always on PostgreSQL) no commit hook runs between
4605 : // groups: close the group's db_version here, or every source db_version would get
4606 : // the same local one and their seqs would collide. A no-op after the RELEASE above.
4607 57902 : if (db_version_changed) cloudsync_pending_version_close(data);
4608 :
4609 57902 : if (!in_savepoint && db_version_changed && !database_in_transaction(data)) {
4610 6855 : rc = database_begin_savepoint(data, "cloudsync_payload_apply");
4611 6855 : if (rc != DBRES_OK) {
4612 0 : cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to start a transaction", rc);
4613 0 : goto cleanup;
4614 : }
4615 6855 : in_savepoint = true;
4616 6855 : applied_at_savepoint = applied;
4617 6855 : }
4618 :
4619 : // Track db_version for batch-flush boundary detection
4620 57902 : if (db_version_changed) {
4621 7005 : last_payload_db_version = decoded_context.db_version;
4622 7005 : }
4623 :
4624 : // Update PK/table tracking
4625 57902 : last_pk = decoded_context.pk;
4626 57902 : last_pk_len = decoded_context.pk_len;
4627 57902 : last_tbl = decoded_context.tbl;
4628 57902 : last_tbl_len = decoded_context.tbl_len;
4629 :
4630 57902 : if (batch.count > 0 && cloudsync_payload_row_is_block(&decoded_context)) {
4631 : // Materialization needs the ordinary columns to exist, but they and
4632 : // every block still belong to the same atomic PK group. Keep its
4633 : // savepoint open until the actual PK/table/db_version boundary.
4634 321 : fail_rc = merge_flush_pending(data);
4635 321 : if (fail_rc != DBRES_OK) break;
4636 321 : }
4637 :
4638 57902 : fail_rc = cloudsync_payload_group_open(data, &batch);
4639 57902 : if (fail_rc != DBRES_OK) break;
4640 57902 : int step_rc = cloudsync_payload_apply_row(data, vm);
4641 57902 : buffer += seek;
4642 57902 : buf_len -= seek;
4643 57902 : dbvm_reset(vm);
4644 57902 : if (step_rc != DBRES_DONE) {
4645 122 : fail_rc = step_rc;
4646 122 : break;
4647 : }
4648 57780 : }
4649 :
4650 : // Close the last PK group: flushed on success, rolled back after a failure.
4651 1782 : if (fail_rc == DBRES_OK) {
4652 1657 : fail_rc = cloudsync_payload_group_flush(data, &batch);
4653 1657 : if (fail_rc == DBRES_OK) applied = (int)nrows;
4654 1657 : } else {
4655 125 : cloudsync_payload_group_abandon(data, &batch);
4656 : }
4657 1782 : data->pending_batch = NULL;
4658 :
4659 1782 : char fail_message[1024] = {0};
4660 1782 : int fail_sqlstate = 0;
4661 1782 : if (fail_rc != DBRES_OK) {
4662 134 : snprintf(fail_message, sizeof(fail_message), "%s", cloudsync_errmsg(data));
4663 134 : fail_sqlstate = cloudsync_sqlstate(data);
4664 134 : }
4665 :
4666 : // Released after a failure too: the failed group is already rolled back, and the
4667 : // groups before it are kept. The receive checkpoint does not move, so they are
4668 : // delivered again and re-merge as no-ops.
4669 1782 : if (in_savepoint) {
4670 1634 : int rc1 = database_commit_savepoint(data, "cloudsync_payload_apply");
4671 1634 : if (rc1 != DBRES_OK) {
4672 80 : applied = applied_at_savepoint;
4673 80 : if (fail_rc == DBRES_OK) {
4674 80 : fail_rc = rc1;
4675 80 : snprintf(fail_message, sizeof(fail_message), "%s", cloudsync_errmsg(data));
4676 80 : fail_sqlstate = cloudsync_sqlstate(data);
4677 80 : }
4678 80 : } else {
4679 1554 : in_savepoint = false;
4680 : }
4681 1634 : }
4682 : // Local writes that follow in the same transaction take a db_version of their own.
4683 1782 : cloudsync_pending_version_close(data);
4684 :
4685 1782 : rc = fail_rc;
4686 3564 : if (rc != DBRES_OK) {
4687 : // The captured message already carries the database error; clear it first so
4688 : // it is not appended to itself a second time.
4689 214 : cloudsync_reset_error(data);
4690 214 : cloudsync_set_error(data, fail_message[0] ? fail_message : "Unable to apply payload changes", rc);
4691 214 : cloudsync_set_sqlstate(data, fail_sqlstate);
4692 214 : } else {
4693 : // Record the last applied (db_version, seq) and advance the receive cursor
4694 : // once, gated on the caller-supplied checkpoint. A non-final chunk passes
4695 : // CLOUDSYNC_CHECKPOINT_NONE so the cursor never lands mid-db_version.
4696 1568 : if (decoded_context.db_version > data->apply_last_db_version ||
4697 0 : (decoded_context.db_version == data->apply_last_db_version && decoded_context.seq > data->apply_last_seq)) {
4698 1568 : data->apply_last_db_version = decoded_context.db_version;
4699 1568 : data->apply_last_seq = decoded_context.seq;
4700 1568 : }
4701 1568 : rc = cloudsync_payload_apply_checkpoint(data, checkpoint_db_version, checkpoint_seq);
4702 : }
4703 :
4704 : cleanup:
4705 : // A failed RELEASE may leave our transaction open. in_savepoint is only set
4706 : // when apply began the transaction from autocommit mode, so a full rollback
4707 : // here cannot discard a caller-owned transaction. ROLLBACK TO plus RELEASE
4708 : // is insufficient: RELEASE can still hit SQLITE_BUSY while ending the write
4709 : // transaction. Preserve the original error across cleanup.
4710 1832 : if (rc != DBRES_OK) {
4711 : char message[1024];
4712 265 : snprintf(message, sizeof(message), "%s", cloudsync_errmsg(data));
4713 265 : int sqlstate = cloudsync_sqlstate(data);
4714 265 : cloudsync_payload_group_abandon(data, &batch);
4715 265 : if (in_savepoint) {
4716 130 : applied = applied_at_savepoint;
4717 130 : if (database_in_transaction(data))
4718 130 : database_exec(data, "ROLLBACK");
4719 130 : }
4720 265 : cloudsync_reset_error(data);
4721 265 : cloudsync_set_error(data, message[0] ? message : "Unable to apply payload changes", rc);
4722 265 : cloudsync_set_sqlstate(data, sqlstate);
4723 265 : }
4724 1832 : data->pending_batch = NULL;
4725 1832 : merge_pending_free_entries(&batch);
4726 1832 : cloudsync_apply_stats_add(data, applied);
4727 :
4728 : // cleanup merge_pending_batch
4729 1832 : if (batch.cached_vm) { databasevm_finalize(batch.cached_vm); batch.cached_vm = NULL; }
4730 1832 : if (batch.cached_col_names) { cloudsync_memory_free(batch.cached_col_names); batch.cached_col_names = NULL; }
4731 1832 : if (batch.entries) { cloudsync_memory_free(batch.entries); batch.entries = NULL; }
4732 :
4733 : // cleanup vm
4734 1832 : if (vm) databasevm_finalize(vm);
4735 :
4736 : // cleanup memory
4737 1832 : if (clone) cloudsync_memory_free(clone);
4738 :
4739 : // error already saved in (save last error)
4740 1832 : if (rc != DBRES_OK) return rc;
4741 :
4742 : // return the number of processed rows
4743 1567 : if (pnrows) *pnrows = nrows;
4744 1567 : return DBRES_OK;
4745 1923 : }
4746 :
4747 : /* Shared by both backends: failures must abort the caller's statement. */
4748 960 : int local_block_update(cloudsync_context *data, cloudsync_table_context *table,
4749 : const void *pk, size_t pklen, int column, const char *text,
4750 : int64_t version, bool initial) {
4751 960 : int rc = DBRES_NOMEM;
4752 960 : const char *col = table_colname(table, column);
4753 960 : block_list_t *old = block_list_create_empty();
4754 960 : block_list_t *next = (text || initial) ? block_split(text ? text : "", table_col_delimiter(table, column)) : block_list_create_empty();
4755 : // Blocks already stored for this column. An update diffs against them; a write of a
4756 : // whole row keeps them out of the diff — it rewrites the column from its first
4757 : // position — and retires below whatever the new value does not cover. A row can hold
4758 : // them at that point because deleting a row keeps its block values, and SQLite's
4759 : // INSERT OR REPLACE skips the delete trigger altogether.
4760 960 : block_list_t *stored = block_list_create_empty();
4761 960 : block_diff_t *diff = NULL;
4762 960 : const char **parts = NULL;
4763 960 : dbvm_t *vm = NULL;
4764 960 : char *sql = NULL;
4765 960 : if (!old || !next || !stored) goto done;
4766 : #ifdef CLOUDSYNC_POSTGRESQL_BUILD
4767 : sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=$1 ORDER BY col_name COLLATE \"C\"", table_blocks_ref(table));
4768 : #else
4769 960 : sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=?1 ORDER BY col_name COLLATE BINARY", table_blocks_ref(table));
4770 : #endif
4771 960 : if (!sql) goto done;
4772 960 : rc = databasevm_prepare(data, sql, &vm, 0);
4773 960 : if (rc != DBRES_OK) goto done;
4774 960 : rc = databasevm_bind_blob(vm, 1, pk, (int)pklen);
4775 960 : if (rc != DBRES_OK) goto done;
4776 5251 : while ((rc = databasevm_step(vm)) == DBRES_ROW) {
4777 4291 : const char *name = database_column_text(vm, 0);
4778 4291 : const char *value = database_column_text(vm, 1);
4779 4291 : const char *pos = block_extract_position_id(name);
4780 : /* Literal prefix comparison: SQL LIKE would mix columns containing % or _. */
4781 4291 : if (pos && (size_t)(pos - name - 1) == strlen(col) && memcmp(name, col, strlen(col)) == 0) {
4782 2979 : if (!block_list_add(initial ? stored : old, value ? value : "", pos)) { rc = DBRES_NOMEM; goto done; }
4783 2979 : }
4784 : }
4785 960 : if (rc != DBRES_DONE) goto done;
4786 960 : databasevm_finalize(vm);
4787 960 : vm = NULL;
4788 960 : rc = DBRES_NOMEM;
4789 960 : if (next->count) {
4790 959 : parts = cloudsync_memory_alloc((uint64_t)next->count * sizeof(*parts));
4791 959 : if (!parts) goto done;
4792 5118 : for (int i = 0; i < next->count; i++) parts[i] = next->entries[i].content;
4793 959 : }
4794 960 : diff = block_diff(old->entries, old->count, parts, next->count);
4795 960 : if (!diff) goto done;
4796 960 : rc = DBRES_OK;
4797 3522 : for (int i = 0; i < diff->count; i++) {
4798 2566 : block_diff_entry_t *entry = &diff->entries[i];
4799 2566 : char *name = block_build_colname(col, entry->position_id);
4800 2566 : if (!name) { rc = DBRES_NOMEM; break; }
4801 2566 : if (entry->type == BLOCK_DIFF_REMOVED) {
4802 81 : rc = local_mark_delete_block_meta(table, pk, pklen, name, version, cloudsync_bumpseq(data));
4803 81 : if (rc == DBRES_OK) rc = block_delete_value_external(data, table, pk, pklen, name);
4804 81 : } else {
4805 2485 : rc = local_mark_insert_or_update_meta(table, pk, pklen, name, version, cloudsync_bumpseq(data));
4806 2485 : dbvm_t *write = table_block_value_write_stmt(table);
4807 2485 : if (rc == DBRES_OK && !write) rc = DBRES_MISUSE;
4808 2485 : if (rc == DBRES_OK) rc = databasevm_bind_blob(write, 1, pk, (int)pklen);
4809 2485 : if (rc == DBRES_OK) rc = databasevm_bind_text(write, 2, name, -1);
4810 2485 : if (rc == DBRES_OK) rc = databasevm_bind_text(write, 3, entry->content, -1);
4811 2485 : if (rc == DBRES_OK) rc = databasevm_step(write);
4812 2485 : if (write) databasevm_reset(write);
4813 2485 : if (rc == DBRES_DONE) rc = DBRES_OK;
4814 : }
4815 2566 : cloudsync_memory_free(name);
4816 2566 : if (rc != DBRES_OK) break;
4817 2562 : }
4818 : // Retire the stored blocks the new value left untouched: their positions were not
4819 : // rewritten above, so they would stay live here and reach the peers as content.
4820 2179 : for (int i = 0; rc == DBRES_OK && i < stored->count; i++) {
4821 1219 : const char *pos = stored->entries[i].position_id;
4822 1219 : bool rewritten = false;
4823 3058 : for (int j = 0; j < diff->count && !rewritten; j++) {
4824 3678 : rewritten = (diff->entries[j].type != BLOCK_DIFF_REMOVED &&
4825 1839 : strcmp(diff->entries[j].position_id, pos) == 0);
4826 1839 : }
4827 1219 : if (rewritten) continue;
4828 301 : char *name = block_build_colname(col, pos);
4829 301 : if (!name) { rc = DBRES_NOMEM; break; }
4830 301 : rc = local_mark_delete_block_meta(table, pk, pklen, name, version, cloudsync_bumpseq(data));
4831 301 : if (rc == DBRES_OK) rc = block_delete_value_external(data, table, pk, pklen, name);
4832 301 : cloudsync_memory_free(name);
4833 1261 : }
4834 : done:
4835 960 : if (vm) databasevm_finalize(vm);
4836 960 : cloudsync_memory_free(sql);
4837 960 : cloudsync_memory_free((void *)parts);
4838 960 : block_diff_free(diff);
4839 960 : block_list_free(old);
4840 960 : block_list_free(next);
4841 960 : block_list_free(stored);
4842 960 : if (rc != DBRES_OK) {
4843 : char message[512];
4844 4 : snprintf(message, sizeof(message), "Unable to write the blocks of column \"%s\" of table \"%s\"", col, table->name);
4845 4 : cloudsync_set_error(data, message, rc);
4846 4 : }
4847 960 : return rc;
4848 : }
4849 :
4850 702 : int local_block_insert(cloudsync_context *data, cloudsync_table_context *table,
4851 : const void *pk, size_t pklen, int column, int64_t version) {
4852 702 : const char *colname = table_colname(table, column);
4853 702 : dbvm_t *vm = table_column_lookup(table, colname, false, NULL);
4854 702 : if (!vm) return cloudsync_set_error(data, "Missing block column statement", DBRES_MISUSE);
4855 :
4856 : // Every failure below must carry a message: databasevm_step clears the error on
4857 : // entry, so a bare code reaches the caller as a blank "not an error".
4858 : char errmsg[512];
4859 702 : int rc = pk_decode_prikey((char *)pk, pklen, pk_decode_bind_callback, vm);
4860 702 : if (rc < 0) {
4861 0 : databasevm_reset(vm);
4862 0 : snprintf(errmsg, sizeof(errmsg), "Unable to decode the primary key of a row in \"%s\" while writing block column \"%s\"", table->name, colname);
4863 0 : return cloudsync_set_error(data, errmsg, DBRES_ERROR);
4864 : }
4865 :
4866 702 : rc = databasevm_step(vm);
4867 702 : char *copy = NULL;
4868 702 : if (rc == DBRES_ROW) {
4869 702 : const char *text = database_column_text(vm, 0);
4870 702 : copy = cloudsync_string_dup(text ? text : "");
4871 702 : if (copy) rc = DBRES_OK;
4872 : else {
4873 0 : snprintf(errmsg, sizeof(errmsg), "Not enough memory to read block column \"%s\" of table \"%s\"", colname, table->name);
4874 0 : rc = cloudsync_set_error(data, errmsg, DBRES_NOMEM);
4875 : }
4876 702 : }
4877 0 : else if (rc == DBRES_DONE) {
4878 : // Reading the row back is part of the block write: without its text there is
4879 : // nothing to split. A row that its own session cannot select cannot sync at
4880 : // all, so report it here rather than leave the column silently untracked.
4881 0 : snprintf(errmsg, sizeof(errmsg), "Unable to read back block column \"%s\" of table \"%s\": the row just written is not visible to this connection (check the table's row-level security SELECT policy)", colname, table->name);
4882 0 : rc = cloudsync_set_error(data, errmsg, DBRES_ERROR);
4883 0 : }
4884 : // End the read cursor before writes that can invoke nested triggers/SPI errors.
4885 702 : databasevm_reset(vm);
4886 702 : if (rc == DBRES_OK) rc = local_block_update(data, table, pk, pklen, column, copy, version, true);
4887 702 : cloudsync_memory_free(copy);
4888 702 : return rc;
4889 702 : }
4890 :
4891 : // MARK: - Payload load/store -
4892 :
4893 0 : int cloudsync_payload_get (cloudsync_context *data, char **blob, int *blob_size, int *db_version, int64_t *new_db_version) {
4894 : // retrieve current db_version and seq
4895 0 : *db_version = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_SEND_DBVERSION);
4896 0 : if (*db_version < 0) return DBRES_ERROR;
4897 :
4898 : // retrieve BLOB
4899 : char sql[1024];
4900 0 : snprintf(sql, sizeof(sql), "WITH max_db_version AS (SELECT MAX(db_version) AS max_db_version FROM cloudsync_changes WHERE site_id=cloudsync_siteid()) "
4901 : "SELECT * FROM (SELECT cloudsync_payload_encode(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq) AS payload, max_db_version AS max_db_version FROM cloudsync_changes, max_db_version WHERE site_id=cloudsync_siteid() AND db_version>%d) WHERE payload IS NOT NULL", *db_version);
4902 :
4903 0 : int64_t len = 0;
4904 0 : int rc = database_select_blob_int(data, sql, blob, &len, new_db_version);
4905 0 : *blob_size = (int)len;
4906 0 : if (rc != DBRES_OK) return rc;
4907 :
4908 : // exit if there is no data to send
4909 0 : if (*blob == NULL || *blob_size == 0) return DBRES_OK;
4910 0 : return rc;
4911 0 : }
4912 :
4913 : #ifdef CLOUDSYNC_DESKTOP_OS
4914 0 : int cloudsync_payload_save (cloudsync_context *data, const char *payload_path, int *size) {
4915 : DEBUG_FUNCTION("cloudsync_payload_save");
4916 :
4917 : // silently delete any other payload with the same name
4918 0 : cloudsync_file_delete(payload_path);
4919 :
4920 : // retrieve payload
4921 0 : char *blob = NULL;
4922 0 : int blob_size = 0, db_version = 0;
4923 0 : int64_t new_db_version = 0;
4924 0 : int rc = cloudsync_payload_get(data, &blob, &blob_size, &db_version, &new_db_version);
4925 0 : if (rc != DBRES_OK) {
4926 0 : if (db_version < 0) return cloudsync_set_error(data, "Unable to retrieve db_version", rc);
4927 0 : return cloudsync_set_error(data, "Unable to retrieve changes in cloudsync_payload_save", rc);
4928 : }
4929 :
4930 : // exit if there is no data to save
4931 0 : if (blob == NULL || blob_size == 0) {
4932 0 : if (size) *size = 0;
4933 0 : return DBRES_OK;
4934 : }
4935 :
4936 : // write payload to file
4937 0 : bool res = cloudsync_file_write(payload_path, blob, (size_t)blob_size);
4938 0 : cloudsync_memory_free(blob);
4939 0 : if (res == false) {
4940 0 : return cloudsync_set_error(data, "Unable to write payload to file path", DBRES_IOERR);
4941 : }
4942 :
4943 : // returns blob size
4944 0 : if (size) *size = blob_size;
4945 0 : return DBRES_OK;
4946 0 : }
4947 : #endif
4948 :
4949 : // MARK: - Core -
4950 :
4951 831 : int cloudsync_table_sanity_check (cloudsync_context *data, const char *name, CLOUDSYNC_INIT_FLAG init_flags) {
4952 : DEBUG_DBFUNCTION("cloudsync_table_sanity_check %s", name);
4953 : char buffer[2048];
4954 :
4955 : // sanity check table name
4956 831 : if (name == NULL) {
4957 1 : return cloudsync_set_error(data, "cloudsync_init requires a non-null table parameter", DBRES_ERROR);
4958 : }
4959 :
4960 : // avoid allocating heap memory for SQL statements by setting a maximum length of 512 characters
4961 : // for table names. This limit is reasonable and helps prevent memory management issues.
4962 830 : const size_t maxlen = CLOUDSYNC_MAX_TABLENAME_LEN;
4963 830 : if (strlen(name) > maxlen) {
4964 1 : snprintf(buffer, sizeof(buffer), "Table name cannot be longer than %d characters", (int)maxlen);
4965 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4966 : }
4967 :
4968 : // check if already initialized
4969 829 : cloudsync_table_context *table = table_lookup(data, name);
4970 829 : if (table) return DBRES_OK;
4971 :
4972 : // check if table exists
4973 824 : if (database_table_exists(data, name, cloudsync_schema(data)) == false) {
4974 2 : snprintf(buffer, sizeof(buffer), "Table %s does not exist", name);
4975 2 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4976 : }
4977 :
4978 : // no more than 128 columns can be used as a composite primary key (SQLite hard limit)
4979 822 : int npri_keys = database_count_pk(data, name, false, cloudsync_schema(data));
4980 822 : if (npri_keys < 0) return cloudsync_set_dberror(data);
4981 822 : if (npri_keys > 128) return cloudsync_set_error(data, "No more than 128 columns can be used to form a composite primary key", DBRES_ERROR);
4982 :
4983 : #if CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
4984 : // if count == 0 means that rowid will be used as primary key (BTW: very bad choice for the user)
4985 822 : if (npri_keys == 0) {
4986 1 : snprintf(buffer, sizeof(buffer), "Rowid only tables are not supported, all primary keys must be explicitly set and declared as NOT NULL (table %s)", name);
4987 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4988 : }
4989 : #endif
4990 :
4991 821 : bool skip_int_pk_check = (init_flags & CLOUDSYNC_INIT_FLAG_SKIP_INT_PK_CHECK) != 0;
4992 821 : if (!skip_int_pk_check) {
4993 758 : if (npri_keys == 1) {
4994 : // the affinity of a column is determined by the declared type of the column,
4995 : // according to the following rules in the order shown:
4996 : // 1. If the declared type contains the string "INT" then it is assigned INTEGER affinity.
4997 670 : int npri_keys_int = database_count_int_pk(data, name, cloudsync_schema(data));
4998 670 : if (npri_keys_int < 0) return cloudsync_set_dberror(data);
4999 670 : if (npri_keys == npri_keys_int) {
5000 1 : snprintf(buffer, sizeof(buffer), "Table %s uses a single-column INTEGER primary key. For CRDT replication, primary keys must be globally unique. Consider using a TEXT primary key with UUIDs or ULID to avoid conflicts across nodes. If you understand the risk and still want to use this INTEGER primary key, set the third argument of the cloudsync_init function to 1 to skip this check.", name);
5001 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
5002 : }
5003 :
5004 669 : }
5005 757 : }
5006 :
5007 : // if user declared explicit primary key(s) then make sure they are all declared as NOT NULL
5008 : #if CLOUDSYNC_CHECK_NOTNULL_PRIKEYS
5009 : bool skip_notnull_prikeys_check = (init_flags & CLOUDSYNC_INIT_FLAG_SKIP_NOT_NULL_PRIKEYS_CHECK) != 0;
5010 : if (!skip_notnull_prikeys_check) {
5011 : if (npri_keys > 0) {
5012 : int npri_keys_notnull = database_count_pk(data, name, true, cloudsync_schema(data));
5013 : if (npri_keys_notnull < 0) return cloudsync_set_dberror(data);
5014 : if (npri_keys != npri_keys_notnull) {
5015 : snprintf(buffer, sizeof(buffer), "All primary keys must be explicitly declared as NOT NULL (table %s)", name);
5016 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
5017 : }
5018 : }
5019 : }
5020 : #endif
5021 :
5022 : // check for columns declared as NOT NULL without a DEFAULT value.
5023 : // Otherwise, col_merge_stmt would fail if changes to other columns are inserted first.
5024 820 : bool skip_notnull_default_check = (init_flags & CLOUDSYNC_INIT_FLAG_SKIP_NOT_NULL_DEFAULT_CHECK) != 0;
5025 820 : if (!skip_notnull_default_check) {
5026 820 : int n_notnull_nodefault = database_count_notnull_without_default(data, name, cloudsync_schema(data));
5027 820 : if (n_notnull_nodefault < 0) return cloudsync_set_dberror(data);
5028 820 : if (n_notnull_nodefault > 0) {
5029 0 : snprintf(buffer, sizeof(buffer), "All non-primary key columns declared as NOT NULL must have a DEFAULT value. (table %s)", name);
5030 0 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
5031 : }
5032 820 : }
5033 :
5034 820 : return DBRES_OK;
5035 831 : }
5036 :
5037 4 : int cloudsync_cleanup_internal (cloudsync_context *data, cloudsync_table_context *table) {
5038 4 : if (cloudsync_context_init(data) == NULL) return DBRES_MISUSE;
5039 :
5040 : // drop meta-table
5041 4 : const char *table_name = table->name;
5042 4 : char *sql = cloudsync_memory_mprintf(SQL_DROP_CLOUDSYNC_TABLE, table->meta_ref);
5043 4 : int rc = database_exec(data, sql);
5044 4 : cloudsync_memory_free(sql);
5045 4 : if (rc != DBRES_OK) {
5046 : char buffer[1024];
5047 0 : snprintf(buffer, sizeof(buffer), "Unable to drop cloudsync table %s_cloudsync in cloudsync_cleanup", table_name);
5048 0 : return cloudsync_set_error(data, buffer, rc);
5049 : }
5050 :
5051 : // drop blocks table if this table has block LWW columns
5052 4 : if (table->blocks_ref) {
5053 1 : sql = cloudsync_memory_mprintf(SQL_DROP_CLOUDSYNC_TABLE, table->blocks_ref);
5054 1 : rc = database_exec(data, sql);
5055 1 : cloudsync_memory_free(sql);
5056 1 : if (rc != DBRES_OK) {
5057 : char buffer[1024];
5058 0 : snprintf(buffer, sizeof(buffer), "Unable to drop blocks table %s_cloudsync_blocks in cloudsync_cleanup", table_name);
5059 0 : return cloudsync_set_error(data, buffer, rc);
5060 : }
5061 1 : }
5062 :
5063 : // drop original triggers
5064 4 : rc = database_delete_triggers(data, table_name);
5065 4 : if (rc != DBRES_OK) {
5066 : char buffer[1024];
5067 0 : snprintf(buffer, sizeof(buffer), "Unable to delete triggers for table %s", table_name);
5068 0 : return cloudsync_set_error(data, buffer, rc);
5069 : }
5070 :
5071 : // remove all table related settings
5072 4 : dbutils_table_settings_set_key_value(data, table_name, NULL, NULL, NULL);
5073 4 : return DBRES_OK;
5074 4 : }
5075 :
5076 4 : int cloudsync_cleanup (cloudsync_context *data, const char *table_name) {
5077 4 : cloudsync_table_context *table = table_lookup(data, table_name);
5078 4 : if (!table) return DBRES_OK;
5079 :
5080 : // TODO: check what happen if cloudsync_cleanup_internal failes (not eveything dropped) and the table is still in memory?
5081 :
5082 4 : int rc = cloudsync_cleanup_internal(data, table);
5083 4 : if (rc != DBRES_OK) return rc;
5084 :
5085 4 : int counter = table_remove(data, table);
5086 4 : table_free(table);
5087 :
5088 4 : if (counter == 0) {
5089 : // cleanup database on last table
5090 2 : cloudsync_reset_siteid(data);
5091 2 : dbutils_settings_cleanup(data);
5092 2 : } else {
5093 2 : if (database_internal_table_exists(data, CLOUDSYNC_TABLE_SETTINGS_NAME) == true) {
5094 2 : cloudsync_update_schema_hash(data);
5095 2 : }
5096 : }
5097 :
5098 4 : return DBRES_OK;
5099 4 : }
5100 :
5101 0 : int cloudsync_cleanup_all (cloudsync_context *data) {
5102 0 : return database_cleanup(data);
5103 : }
5104 :
5105 1562 : int cloudsync_terminate (cloudsync_context *data) {
5106 : // can't use for/loop here because data->tables_count is changed by table_remove
5107 2358 : while (data->tables_count > 0) {
5108 796 : cloudsync_table_context *t = data->tables[data->tables_count - 1];
5109 796 : table_remove(data, t);
5110 796 : table_free(t);
5111 : }
5112 :
5113 1562 : if (data->schema_version_stmt) databasevm_finalize(data->schema_version_stmt);
5114 1562 : if (data->data_version_stmt) databasevm_finalize(data->data_version_stmt);
5115 1562 : if (data->db_version_stmt) databasevm_finalize(data->db_version_stmt);
5116 1562 : if (data->getset_siteid_stmt) databasevm_finalize(data->getset_siteid_stmt);
5117 1562 : if (data->current_schema) cloudsync_memory_free(data->current_schema);
5118 :
5119 1562 : data->schema_version_stmt = NULL;
5120 1562 : data->data_version_stmt = NULL;
5121 1562 : data->db_version_stmt = NULL;
5122 1562 : data->getset_siteid_stmt = NULL;
5123 1562 : data->current_schema = NULL;
5124 :
5125 : // reset the site_id so the cloudsync_context_init will be executed again
5126 : // if any other cloudsync function is called after terminate
5127 1562 : data->site_id[0] = 0;
5128 :
5129 1562 : return 1;
5130 : }
5131 :
5132 826 : int cloudsync_init_table (cloudsync_context *data, const char *table_name, const char *algo_name, CLOUDSYNC_INIT_FLAG init_flags) {
5133 : // sanity check table and its primary key(s)
5134 826 : int rc = cloudsync_table_sanity_check(data, table_name, init_flags);
5135 826 : if (rc != DBRES_OK) return rc;
5136 :
5137 : // init cloudsync_settings
5138 824 : if (cloudsync_context_init(data) == NULL) {
5139 0 : return cloudsync_set_error(data, "Unable to initialize cloudsync context", DBRES_MISUSE);
5140 : }
5141 :
5142 : // sanity check algo name (if exists)
5143 824 : table_algo algo_new = table_algo_none;
5144 824 : if (!algo_name) algo_name = CLOUDSYNC_DEFAULT_ALGO;
5145 :
5146 824 : algo_new = cloudsync_algo_from_name(algo_name);
5147 824 : if (algo_new == table_algo_none) {
5148 : char buffer[1024];
5149 1 : snprintf(buffer, sizeof(buffer), "Unknown CRDT algorithm name %s", algo_name);
5150 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
5151 : }
5152 :
5153 : // DWS and AWS algorithms are not yet implemented in the merge logic
5154 823 : if (algo_new == table_algo_crdt_dws || algo_new == table_algo_crdt_aws) {
5155 : char buffer[1024];
5156 2 : snprintf(buffer, sizeof(buffer), "CRDT algorithm %s is not yet supported", algo_name);
5157 2 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
5158 : }
5159 :
5160 : // check if table name was already augmented
5161 821 : table_algo algo_current = dbutils_table_settings_get_algo(data, table_name);
5162 :
5163 : // sanity check algorithm
5164 821 : if ((algo_new == algo_current) && (algo_current != table_algo_none)) {
5165 : // if table algorithms and the same and not none, do nothing
5166 821 : } else if ((algo_new == table_algo_none) && (algo_current == table_algo_none)) {
5167 : // nothing is written into settings because the default table_algo_crdt_cls will be used
5168 0 : algo_new = algo_current = table_algo_crdt_cls;
5169 790 : } else if ((algo_new == table_algo_none) && (algo_current != table_algo_none)) {
5170 : // algo is already written into settins so just use it
5171 0 : algo_new = algo_current;
5172 790 : } else if ((algo_new != table_algo_none) && (algo_current == table_algo_none)) {
5173 : // write table algo name in settings
5174 790 : dbutils_table_settings_set_key_value(data, table_name, "*", "algo", algo_name);
5175 790 : } else {
5176 : // error condition
5177 0 : return cloudsync_set_error(data, "The function cloudsync_cleanup(table) must be called before changing a table algorithm", DBRES_MISUSE);
5178 : }
5179 :
5180 : // Run the following function even if table was already augmented.
5181 : // It is safe to call the following function multiple times, if there is nothing to update nothing will be changed.
5182 : // After an alter table, in contrast, all the cloudsync triggers, tables and stmts must be recreated.
5183 :
5184 : // sync algo with table (unused in this version)
5185 : // cloudsync_sync_table_key(data, table_name, "*", CLOUDSYNC_KEY_ALGO, crdt_algo_name(algo_new));
5186 :
5187 : // read row-level filter from settings (if any)
5188 : char init_filter_buf[2048];
5189 821 : int init_frc = dbutils_table_settings_get_value(data, table_name, "*", "filter", init_filter_buf, sizeof(init_filter_buf));
5190 821 : const char *init_filter = (init_frc == DBRES_OK && init_filter_buf[0]) ? init_filter_buf : NULL;
5191 :
5192 : // check triggers
5193 821 : rc = database_create_triggers(data, table_name, algo_new, init_filter);
5194 821 : if (rc != DBRES_OK) return cloudsync_set_error(data, "An error occurred while creating triggers", DBRES_MISUSE);
5195 :
5196 : // check meta-table
5197 821 : rc = database_create_metatable(data, table_name);
5198 821 : if (rc != DBRES_OK) return cloudsync_set_error(data, "An error occurred while creating metatable", DBRES_MISUSE);
5199 :
5200 : // add prepared statements
5201 821 : if (cloudsync_add_dbvms(data) != DBRES_OK) {
5202 0 : return cloudsync_set_error(data, "An error occurred while trying to compile prepared SQL statements", DBRES_MISUSE);
5203 : }
5204 :
5205 : // add table to in-memory data context
5206 821 : if (table_add_to_context(data, algo_new, table_name) == false) {
5207 : char buffer[1024];
5208 0 : snprintf(buffer, sizeof(buffer), "An error occurred while adding %s table information to global context", table_name);
5209 0 : return cloudsync_set_error(data, buffer, DBRES_MISUSE);
5210 : }
5211 :
5212 821 : if (cloudsync_refill_metatable(data, table_name) != DBRES_OK) {
5213 0 : return cloudsync_set_error(data, "An error occurred while trying to fill the augmented table", DBRES_MISUSE);
5214 : }
5215 :
5216 821 : return DBRES_OK;
5217 826 : }
|
