Skip to content

Commit 4c43ebb

Browse files
andinuxandinux
committed
Merge branch 'main' into dev
2 parents a6f8b1b + cb582c1 commit 4c43ebb

File tree

13 files changed

+337
-132
lines changed

13 files changed

+337
-132
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,3 +50,4 @@ jniLibs/
5050
.DS_Store
5151
Thumbs.db
5252
CLAUDE.md
53+
*.o

‎src/cloudsync.c‎

Lines changed: 47 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,6 @@ struct cloudsync_context {
115115
void *aux_data;
116116

117117
// stmts and context values
118-
bool pragma_checked; // we need to check PRAGMAs only once per transaction
119118
dbvm_t *schema_version_stmt;
120119
dbvm_t *data_version_stmt;
121120
dbvm_t *db_version_stmt;
@@ -255,13 +254,15 @@ const char *cloudsync_algo_name (table_algo algo) {
255254
// MARK: - DBVM Utils -
256255

257256
DBVM_VALUE dbvm_execute (dbvm_t *stmt, cloudsync_context *data) {
257+
if (!stmt) return DBVM_VALUE_ERROR;
258+
258259
int rc = databasevm_step(stmt);
259260
if (rc != DBRES_ROW && rc != DBRES_DONE) {
260261
if (data) DEBUG_DBERROR(rc, "stmt_execute", data);
261262
databasevm_reset(stmt);
262263
return DBVM_VALUE_ERROR;
263264
}
264-
265+
265266
DBVM_VALUE result = DBVM_VALUE_CHANGED;
266267
if (stmt == data->data_version_stmt) {
267268
int version = (int)database_column_int(stmt, 0);
@@ -365,12 +366,17 @@ int cloudsync_dbversion_rebuild (cloudsync_context *data) {
365366
int cloudsync_dbversion_rerun (cloudsync_context *data) {
366367
DBVM_VALUE schema_changed = dbvm_execute(data->schema_version_stmt, data);
367368
if (schema_changed == DBVM_VALUE_ERROR) return -1;
368-
369+
369370
if (schema_changed == DBVM_VALUE_CHANGED) {
370371
int rc = cloudsync_dbversion_rebuild(data);
371372
if (rc != DBRES_OK) return -1;
372373
}
373-
374+
375+
if (!data->db_version_stmt) {
376+
data->db_version = CLOUDSYNC_MIN_DB_VERSION;
377+
return 0;
378+
}
379+
374380
DBVM_VALUE rc = dbvm_execute(data->db_version_stmt, data);
375381
if (rc == DBVM_VALUE_ERROR) return -1;
376382
return 0;
@@ -559,7 +565,7 @@ void cloudsync_set_auxdata (cloudsync_context *data, void *xdata) {
559565
}
560566

561567
void cloudsync_set_schema (cloudsync_context *data, const char *schema) {
562-
if (data->current_schema == schema) return;
568+
if (data->current_schema && schema && strcmp(data->current_schema, schema) == 0) return;
563569
if (data->current_schema) cloudsync_memory_free(data->current_schema);
564570
data->current_schema = NULL;
565571
if (schema) data->current_schema = cloudsync_string_dup_lowercase(schema);
@@ -748,7 +754,7 @@ int table_add_stmts (cloudsync_table_context *table, int ncols) {
748754
if (rc != DBRES_OK) goto cleanup;
749755

750756
// precompile the insert local sentinel statement
751-
sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_COL_INIT_OR_BUMP_VERSION, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
757+
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);
752758
if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
753759
DEBUG_SQL("meta_sentinel_insert_stmt: %s", sql);
754760

@@ -920,37 +926,44 @@ int table_remove (cloudsync_context *data, cloudsync_table_context *table) {
920926
int table_add_to_context_cb (void *xdata, int ncols, char **values, char **names) {
921927
cloudsync_table_context *table = (cloudsync_table_context *)xdata;
922928
cloudsync_context *data = table->context;
923-
929+
924930
int index = table->ncols;
925931
for (int i=0; i<ncols; i+=2) {
926932
const char *name = values[i];
927933
int cid = (int)strtol(values[i+1], NULL, 0);
928-
934+
929935
table->col_id[index] = cid;
930936
table->col_name[index] = cloudsync_string_dup_lowercase(name);
931-
if (!table->col_name[index]) return 1;
932-
937+
if (!table->col_name[index]) goto error;
938+
933939
char *sql = table_build_mergeinsert_sql(table, name);
934-
if (!sql) return DBRES_NOMEM;
940+
if (!sql) goto error;
935941
DEBUG_SQL("col_merge_stmt[%d]: %s", index, sql);
936-
942+
937943
int rc = databasevm_prepare(data, sql, (void **)&table->col_merge_stmt[index], DBFLAG_PERSISTENT);
938944
cloudsync_memory_free(sql);
939-
if (rc != DBRES_OK) return rc;
940-
if (!table->col_merge_stmt[index]) return DBRES_MISUSE;
941-
945+
if (rc != DBRES_OK) goto error;
946+
if (!table->col_merge_stmt[index]) goto error;
947+
942948
sql = table_build_value_sql(table, name);
943-
if (!sql) return DBRES_NOMEM;
949+
if (!sql) goto error;
944950
DEBUG_SQL("col_value_stmt[%d]: %s", index, sql);
945-
951+
946952
rc = databasevm_prepare(data, sql, (void **)&table->col_value_stmt[index], DBFLAG_PERSISTENT);
947953
cloudsync_memory_free(sql);
948-
if (rc != DBRES_OK) return rc;
949-
if (!table->col_value_stmt[index]) return DBRES_MISUSE;
954+
if (rc != DBRES_OK) goto error;
955+
if (!table->col_value_stmt[index]) goto error;
950956
}
951957
table->ncols += 1;
952-
958+
953959
return 0;
960+
961+
error:
962+
// clean up partially-initialized entry at index
963+
if (table->col_name[index]) {cloudsync_memory_free(table->col_name[index]); table->col_name[index] = NULL;}
964+
if (table->col_merge_stmt[index]) {databasevm_finalize(table->col_merge_stmt[index]); table->col_merge_stmt[index] = NULL;}
965+
if (table->col_value_stmt[index]) {databasevm_finalize(table->col_value_stmt[index]); table->col_value_stmt[index] = NULL;}
966+
return 1;
954967
}
955968

956969
bool table_ensure_capacity (cloudsync_context *data) {
@@ -992,7 +1005,7 @@ bool table_add_to_context (cloudsync_context *data, table_algo algo, const char
9921005
table->npks = count;
9931006
if (table->npks == 0) {
9941007
#if CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
995-
return false;
1008+
goto abort_add_table;
9961009
#else
9971010
table->rowid_only = true;
9981011
table->npks = 1; // rowid
@@ -1039,7 +1052,8 @@ bool table_add_to_context (cloudsync_context *data, table_algo algo, const char
10391052

10401053
dbvm_t *cloudsync_colvalue_stmt (cloudsync_context *data, const char *tbl_name, bool *persistent) {
10411054
dbvm_t *vm = NULL;
1042-
1055+
*persistent = false;
1056+
10431057
cloudsync_table_context *table = table_lookup(data, tbl_name);
10441058
if (table) {
10451059
char *col_name = NULL;
@@ -1082,7 +1096,7 @@ const char *table_colname (cloudsync_table_context *table, int index) {
10821096
bool table_pk_exists (cloudsync_table_context *table, const char *value, size_t len) {
10831097
// check if a row with the same primary key already exists
10841098
// if so, this means the row might have been previously deleted (sentinel)
1085-
return (bool)dbvm_count(table->meta_pkexists_stmt, value, len, DBTYPE_BLOB);
1099+
return (dbvm_count(table->meta_pkexists_stmt, value, len, DBTYPE_BLOB) > 0);
10861100
}
10871101

10881102
char **table_pknames (cloudsync_table_context *table) {
@@ -1373,6 +1387,10 @@ int merge_did_cid_win (cloudsync_context *data, cloudsync_table_context *table,
13731387
rc = databasevm_step(vm);
13741388
if (rc == DBRES_ROW) {
13751389
const void *local_site_id = database_column_blob(vm, 0);
1390+
if (!local_site_id) {
1391+
dbvm_reset(vm);
1392+
return cloudsync_set_error(data, "NULL site_id in cloudsync table, table is probably corrupted", DBRES_ERROR);
1393+
}
13761394
ret = memcmp(site_id, local_site_id, site_len);
13771395
*didwin_flag = (ret > 0);
13781396
dbvm_reset(vm);
@@ -1929,6 +1947,7 @@ int cloudsync_refill_metatable (cloudsync_context *data, const char *table_name)
19291947
rc = databasevm_step(vm);
19301948
if (rc == DBRES_ROW) {
19311949
const char *pk = (const char *)database_column_text(vm, 0);
1950+
if (!pk) { rc = DBRES_ERROR; break; }
19321951
size_t pklen = strlen(pk);
19331952
rc = local_mark_insert_or_update_meta(table, pk, pklen, col_name, db_version, cloudsync_bumpseq(data));
19341953
} else if (rc == DBRES_DONE) {
@@ -2448,8 +2467,8 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
24482467
if (in_savepoint && db_version_changed) {
24492468
rc = database_commit_savepoint(data, "cloudsync_payload_apply");
24502469
if (rc != DBRES_OK) {
2451-
if (clone) cloudsync_memory_free(clone);
2452-
return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to release a savepoint", rc);
2470+
cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to release a savepoint", rc);
2471+
goto cleanup;
24532472
}
24542473
in_savepoint = false;
24552474
}
@@ -2459,8 +2478,8 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
24592478
if (!in_transaction && db_version_changed) {
24602479
rc = database_begin_savepoint(data, "cloudsync_payload_apply");
24612480
if (rc != DBRES_OK) {
2462-
if (clone) cloudsync_memory_free(clone);
2463-
return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to start a transaction", rc);
2481+
cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to start a transaction", rc);
2482+
goto cleanup;
24642483
}
24652484
last_payload_db_version = decoded_context.db_version;
24662485
in_savepoint = true;
@@ -2548,7 +2567,7 @@ int cloudsync_payload_get (cloudsync_context *data, char **blob, int *blob_size,
25482567
if (rc != DBRES_OK) return rc;
25492568

25502569
// exit if there is no data to send
2551-
if (blob == NULL || *blob_size == 0) return DBRES_OK;
2570+
if (*blob == NULL || *blob_size == 0) return DBRES_OK;
25522571
return rc;
25532572
}
25542573

‎src/cloudsync.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717
extern "C" {
1818
#endif
1919

20-
#define CLOUDSYNC_VERSION "0.9.110"
20+
#define CLOUDSYNC_VERSION "0.9.112"
2121
#define CLOUDSYNC_MAX_TABLENAME_LEN 512
2222

2323
#define CLOUDSYNC_VALUE_NOTSET -1

‎src/dbutils.c‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -411,14 +411,16 @@ int dbutils_settings_init (cloudsync_context *data) {
411411
if (rc != DBRES_OK) return rc;
412412

413413
// library version
414-
char sql[1024];
415-
snprintf(sql, sizeof(sql), SQL_INSERT_SETTINGS_STR_FORMAT, CLOUDSYNC_KEY_LIBVERSION, CLOUDSYNC_VERSION);
414+
char *sql = cloudsync_memory_mprintf(SQL_INSERT_SETTINGS_STR_FORMAT, CLOUDSYNC_KEY_LIBVERSION, CLOUDSYNC_VERSION);
415+
if (!sql) return DBRES_NOMEM;
416416
rc = database_exec(data, sql);
417+
cloudsync_memory_free(sql);
417418
if (rc != DBRES_OK) return rc;
418-
419+
419420
// schema version
420-
snprintf(sql, sizeof(sql), SQL_INSERT_SETTINGS_INT_FORMAT, CLOUDSYNC_KEY_SCHEMAVERSION, (long long)database_schema_version(data));
421-
rc = database_exec(data, sql);
421+
char sql_int[1024];
422+
snprintf(sql_int, sizeof(sql_int), SQL_INSERT_SETTINGS_INT_FORMAT, CLOUDSYNC_KEY_SCHEMAVERSION, (long long)database_schema_version(data));
423+
rc = database_exec(data, sql_int);
422424
if (rc != DBRES_OK) return rc;
423425
}
424426

‎src/network.c‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -942,7 +942,7 @@ void cloudsync_network_logout (sqlite3_context *context, int argc, sqlite3_value
942942
}
943943

944944
// run everything in a savepoint
945-
rc = database_begin_savepoint(data, "cloudsync_logout_savepoint;");
945+
rc = database_begin_savepoint(data, "cloudsync_logout_savepoint");
946946
if (rc != SQLITE_OK) {
947947
errmsg = cloudsync_memory_mprintf("Unable to create cloudsync_logout savepoint %s", cloudsync_errmsg(data));
948948
goto finalize;

‎src/postgresql/cloudsync_postgresql.c‎

Lines changed: 31 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -473,7 +473,7 @@ Datum pg_cloudsync_terminate (PG_FUNCTION_ARGS) {
473473
PG_END_TRY();
474474

475475
if (spi_connected) SPI_finish();
476-
PG_RETURN_INT32(rc);
476+
PG_RETURN_BOOL(rc == DBRES_OK);
477477
}
478478

479479
// MARK: - Settings Functions -
@@ -820,8 +820,7 @@ Datum cloudsync_payload_encode_transfn (PG_FUNCTION_ARGS) {
820820
// Get or allocate aggregate state
821821
if (PG_ARGISNULL(0)) {
822822
MemoryContext oldContext = MemoryContextSwitchTo(aggContext);
823-
payload = (cloudsync_payload_context *)cloudsync_memory_alloc(cloudsync_payload_context_size(NULL));
824-
memset(payload, 0, cloudsync_payload_context_size(NULL));
823+
payload = (cloudsync_payload_context *)palloc0(cloudsync_payload_context_size(NULL));
825824
MemoryContextSwitchTo(oldContext);
826825
} else {
827826
payload = (cloudsync_payload_context *)PG_GETARG_POINTER(0);
@@ -1819,13 +1818,16 @@ static Oid get_column_oid(const char *schema, const char *table_name, const char
18191818
pfree(DatumGetPointer(values[1]));
18201819
if (schema) pfree(DatumGetPointer(values[2]));
18211820

1822-
if (ret != SPI_OK_SELECT || SPI_processed == 0) return InvalidOid;
1821+
if (ret != SPI_OK_SELECT || SPI_processed == 0) {
1822+
if (SPI_tuptable) SPI_freetuptable(SPI_tuptable);
1823+
return InvalidOid;
1824+
}
18231825

18241826
bool isnull;
18251827
Datum col_oid = SPI_getbinval(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1, &isnull);
1826-
if (isnull) return InvalidOid;
1827-
1828-
return DatumGetObjectId(col_oid);
1828+
Oid result = isnull ? InvalidOid : DatumGetObjectId(col_oid);
1829+
SPI_freetuptable(SPI_tuptable);
1830+
return result;
18291831
}
18301832

18311833
// Decode encoded bytea into a pgvalue_t with the decoded base type.
@@ -1958,23 +1960,34 @@ Datum cloudsync_col_value(PG_FUNCTION_ARGS) {
19581960
}
19591961

19601962
// execute vm
1961-
Datum d = (Datum)0;
19621963
int rc = databasevm_step(vm);
19631964
if (rc == DBRES_DONE) {
1964-
rc = DBRES_OK;
1965-
PG_RETURN_CSTRING(CLOUDSYNC_RLS_RESTRICTED_VALUE);
1965+
databasevm_reset(vm);
1966+
// row not found (RLS or genuinely missing) — return the RLS sentinel as bytea
1967+
const char *rls = CLOUDSYNC_RLS_RESTRICTED_VALUE;
1968+
size_t rls_len = strlen(rls);
1969+
bytea *result = (bytea *)palloc(VARHDRSZ + rls_len);
1970+
SET_VARSIZE(result, VARHDRSZ + rls_len);
1971+
memcpy(VARDATA(result), rls, rls_len);
1972+
PG_RETURN_BYTEA_P(result);
19661973
} else if (rc == DBRES_ROW) {
1967-
// store value result
1968-
rc = DBRES_OK;
1969-
d = database_column_datum(vm, 0);
1970-
}
1971-
1972-
if (rc != DBRES_OK) {
1974+
// copy value before reset invalidates SPI tuple memory
1975+
const void *blob = database_column_blob(vm, 0);
1976+
int blob_len = database_column_bytes(vm, 0);
1977+
bytea *result = NULL;
1978+
if (blob && blob_len > 0) {
1979+
result = (bytea *)palloc(VARHDRSZ + blob_len);
1980+
SET_VARSIZE(result, VARHDRSZ + blob_len);
1981+
memcpy(VARDATA(result), blob, blob_len);
1982+
}
19731983
databasevm_reset(vm);
1974-
ereport(ERROR, (errmsg("cloudsync_col_value error: %s", cloudsync_errmsg(data))));
1984+
if (result) PG_RETURN_BYTEA_P(result);
1985+
PG_RETURN_NULL();
19751986
}
1987+
19761988
databasevm_reset(vm);
1977-
PG_RETURN_DATUM(d);
1989+
ereport(ERROR, (errmsg("cloudsync_col_value error: %s", cloudsync_errmsg(data))));
1990+
PG_RETURN_NULL(); // unreachable, silences compiler
19781991
}
19791992

19801993
// Track SRF execution state across calls

0 commit comments

Comments
 (0)