diff --git a/.github/workflows/changelog.yml b/.github/workflows/changelog.yml index 107ae0c4..b958e391 100644 --- a/.github/workflows/changelog.yml +++ b/.github/workflows/changelog.yml @@ -3,8 +3,12 @@ name: Release and Update Website Changelog on: push: tags: - # main.yml creates release tags without a "v" prefix (e.g. 1.1.3) - - "*.*.*" + # NOTE: this never matches. main.yml creates release tags without a "v" + # prefix (e.g. 1.1.3), so the changelog is published by running this + # workflow manually. Widening the filter is not enough on its own: the + # called workflow derives the version with ${GITHUB_REF#refs/tags/v} and + # rejects an unprefixed tag. Fixing it needs a change in changelog-action. + - "v*.*.*" workflow_dispatch: inputs: test_version: diff --git a/API.md b/API.md index 9c3419f2..cb77655c 100644 --- a/API.md +++ b/API.md @@ -813,10 +813,11 @@ If the network is misconfigured or the remote server is unreachable, the functio **Returns:** A JSON string with the receive result: ```json -{"receive": {"rows": N, "tables": ["table1", "table2"], "chunks": C, "bytes": B, "complete": true, "error": "...", "lastFailure": {...}}} +{"receive": {"rows": N, "denied": D, "tables": ["table1", "table2"], "chunks": C, "bytes": B, "complete": true, "error": "...", "lastFailure": {...}}} ``` - `receive.rows`: The total number of rows received and applied to the local database, summed across all chunks drained this call. `0` when the receive phase failed, when nothing was available, or when only intermediate fragments were staged without completing a value. +- `receive.denied`: The number of entries a row-level security policy rejected, summed across all chunks drained this call. Denied entries are skipped and the receive cursor still advances past them: the rejection is permanent, so holding the cursor back would re-deliver the same entries on every call. They are not counted in `receive.rows`, so a non-zero `denied` alongside a `rows` of `0` means nothing was written — the shape of an apply connection whose session identity is not set. - `receive.tables`: An array of table names that received changes (the union across all drained chunks). Empty (`[]`) if no changes were applied or the receive phase failed. - `receive.chunks`: The number of payload chunks applied by this call. `0` when nothing was ready, `1` for a single monolithic/inline page, and `N` for a drained `N`-chunk stream (bounded by `max_chunks` if given). - `receive.bytes`: The total serialized payload bytes received this call (uncompressed cloudsync payload size, summed across chunks; transport-independent, not the compressed wire size). Useful for byte-budgeted draining together with `max_chunks`. @@ -828,16 +829,16 @@ If the network is misconfigured or the remote server is unreachable, the functio ```sql SELECT cloudsync_network_receive_changes(); --- '{"receive":{"rows":3,"tables":["tasks"],"chunks":1,"bytes":820,"complete":true}}' +-- '{"receive":{"rows":3,"denied":0,"tables":["tasks"],"chunks":1,"bytes":820,"complete":true}}' -- Capped drain with more pending (call again to continue): --- '{"receive":{"rows":40,"tables":["docs"],"chunks":5,"bytes":1310720,"complete":false}}' +-- '{"receive":{"rows":40,"denied":0,"tables":["docs"],"chunks":5,"bytes":1310720,"complete":false}}' -- With a client-side apply error: --- '{"receive":{"rows":0,"tables":[],"chunks":0,"bytes":0,"complete":true,"error":"Cannot apply the received payload because the schema hash is unknown 7218827471400075525."}}' +-- '{"receive":{"rows":0,"denied":0,"tables":[],"chunks":0,"bytes":0,"complete":true,"error":"Cannot apply the received payload because the schema hash is unknown 7218827471400075525."}}' -- With a server-reported check-job failure: --- '{"receive":{"rows":0,"tables":[],"chunks":0,"bytes":0,"complete":true,"lastFailure":{"jobId":456,"dbVersion":15,"seq":1,"code":"tenant_unreachable","stage":"encode_changes","message":"tenant check failed","retryable":true,"failedAt":"2026-04-24T10:22:00Z"}}}' +-- '{"receive":{"rows":0,"denied":0,"tables":[],"chunks":0,"bytes":0,"complete":true,"lastFailure":{"jobId":456,"dbVersion":15,"seq":1,"code":"tenant_unreachable","stage":"encode_changes","message":"tenant check failed","retryable":true,"failedAt":"2026-04-24T10:22:00Z"}}}' ``` --- @@ -867,7 +868,7 @@ When the server delivers changes as a stream of chunks, this function drains the ```json { "send": {"status": "synced|syncing|out-of-sync|error", "localVersion": N, "serverVersion": N, "chunks": C, "bytes": B, "lastFailure": {...}}, - "receive": {"rows": N, "tables": ["table1", "table2"], "chunks": C, "bytes": B, "complete": true, "error": "...", "lastFailure": {...}} + "receive": {"rows": N, "denied": D, "tables": ["table1", "table2"], "chunks": C, "bytes": B, "complete": true, "error": "...", "lastFailure": {...}} } ``` @@ -877,6 +878,7 @@ When the server delivers changes as a stream of chunks, this function drains the - `send.chunks` / `send.bytes`: Number of payload chunks sent and total serialized payload bytes sent during the send phase. Same semantics as in [`cloudsync_network_send_changes()`](#cloudsync_network_send_changes). - `send.lastFailure` (optional): Same semantics as in [`cloudsync_network_send_changes()`](#cloudsync_network_send_changes) — forwarded verbatim from the server's `failures.apply` whenever a failed apply job is reported, regardless of `status`. - `receive.rows`: The **total** number of rows received and applied during the receive phase, summed across **all** chunks drained in this call. `0` when the receive phase failed. +- `receive.denied`: The **total** number of entries rejected by a row-level security policy across **all** chunks drained in this call. Skipped rather than retried, and not counted in `receive.rows` — see [Receive Changes](#receive-changes). - `receive.tables`: An array of table names that received changes (the union across all drained chunks). Empty (`[]`) if no changes were applied or the receive phase failed. - `receive.chunks`: The number of payload chunks applied in this call. `0` when nothing was ready, `1` for a single monolithic/inline page, and `N` for a fully drained `N`-chunk stream. `cloudsync_network_sync()` always drains the whole stream (it does not cap chunks). - `receive.bytes`: The total serialized payload bytes received this call (uncompressed cloudsync payload size, summed across chunks; not the compressed wire size). Same semantics as in [`cloudsync_network_receive_changes()`](#cloudsync_network_receive_changesmax_chunks). @@ -889,15 +891,15 @@ When the server delivers changes as a stream of chunks, this function drains the ```sql -- Perform a single synchronization cycle SELECT cloudsync_network_sync(); --- '{"send":{"status":"synced","localVersion":5,"serverVersion":5,"chunks":1,"bytes":2048},"receive":{"rows":3,"tables":["tasks"],"chunks":1,"bytes":820,"complete":true}}' +-- '{"send":{"status":"synced","localVersion":5,"serverVersion":5,"chunks":1,"bytes":2048},"receive":{"rows":3,"denied":0,"tables":["tasks"],"chunks":1,"bytes":820,"complete":true}}' -- Perform a synchronization cycle with custom retry settings SELECT cloudsync_network_sync(500, 3); -- A large download drained as a multi-chunk stream in a single call: --- '{"send":{"status":"synced","localVersion":42,"serverVersion":42,"chunks":0,"bytes":0},"receive":{"rows":1200,"tables":["docs"],"chunks":7,"bytes":1835008,"complete":true}}' +-- '{"send":{"status":"synced","localVersion":42,"serverVersion":42,"chunks":0,"bytes":0},"receive":{"rows":1200,"denied":0,"tables":["docs"],"chunks":7,"bytes":1835008,"complete":true}}' -- Receive phase failed but send phase completed — the error is surfaced in JSON, not as a SQL error: --- '{"send":{"status":"synced","localVersion":5,"serverVersion":5,"chunks":1,"bytes":512},"receive":{"rows":0,"tables":[],"chunks":0,"bytes":0,"complete":false,"error":"Cannot apply the received payload because the schema hash is unknown 7218827471400075525."}}' +-- '{"send":{"status":"synced","localVersion":5,"serverVersion":5,"chunks":1,"bytes":512},"receive":{"rows":0,"denied":0,"tables":[],"chunks":0,"bytes":0,"complete":false,"error":"Cannot apply the received payload because the schema hash is unknown 7218827471400075525."}}' ``` --- diff --git a/CHANGELOG.md b/CHANGELOG.md index 44d16fc7..c4426fec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,24 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). +## [1.1.4] - 2026-09-11 + +### Added + +- **Rows rejected by a row-level security policy are now reported** as `receive.denied` in the JSON returned by `cloudsync_network_receive_changes()` and `cloudsync_network_sync()`, counted across every chunk of a receive. A denial is a permanent, expected outcome — those rows are not this site's to hold — so they are skipped and the receive cursor still advances past them; without a count, discarding them would be invisible. They are not counted in `receive.rows`, which reports what was actually written. This does not yet extend to a value large enough to be sent in fragments: a denial there is still reported as a receive error and the cursor does not advance. A non-zero `denied` with a zero `rows` is the shape of an apply connection whose session identity (`auth.uid()` / `app.current_user_id`) is not set. +- **Network requests now have deadlines**, where previously a stalled server left a sync call waiting indefinitely. 30 seconds to connect, on every request. API calls are then capped at 300 seconds of elapsed time. Artifact transfers are bounded on progress instead — they abort after 60 seconds below 1 KB/s — because a large payload on a slow link would otherwise be killed mid-flight by an elapsed-time cap, and because a genuine stall is detected sooner this way. A 1-hour absolute backstop still bounds an artifact transfer that trickles just fast enough to stay alive. Override at build time with `CLOUDSYNC_CONNECT_TIMEOUT_SECONDS`, `CLOUDSYNC_REQUEST_TIMEOUT_SECONDS`, `CLOUDSYNC_ARTIFACT_LOW_SPEED_LIMIT`, `CLOUDSYNC_ARTIFACT_LOW_SPEED_TIME` and `CLOUDSYNC_ARTIFACT_TIMEOUT_SECONDS`. +- **Decompressed payloads are capped at 256 MiB before allocation**, overridable with `CLOUDSYNC_MAX_PAYLOAD_EXPANDED_SIZE` (LZ4's own `INT_MAX` bound still applies). Default chunk sizes are far below this limit; an oversized legacy monolithic payload must be rechunked or the limit raised explicitly. + +### Fixed + +- **A failed payload write now reports the error and leaves the receive cursor where it was.** An error on one row could previously be overwritten by a later successful row, so `cloudsync_payload_apply` could report success after dropping changes and still advance the checkpoint — losing them silently. SQLite keeps its existing per-group partial-application behaviour. +- **Primary-key doubles keep their deployed little-endian IEEE754 byte order on every architecture.** The previous code combined host conversion with manual big-endian serialization; the historical format is now explicit. Integer keys are unchanged, and no migration is needed for little-endian deployments. +- **PostgreSQL no longer frees a tuple table belonging to another open cursor.** A block write that failed while a second SPI cursor was active could release rows still in use; tuple tables are now owned per statement. +- **Block-level LWW text writes roll back cleanly when a block write fails**, instead of leaving the row partially written. +- **Block-column failures now say which table and column failed**, instead of aborting the statement with a blank message. Reading the row back is part of writing a block column, so a row the session cannot `SELECT` — a row-level security policy narrower for reads than for writes, say — is reported with that cause rather than as an empty "not an error". +- **64-bit clock values above `UINT32_MAX` are handled correctly on incoming changes** — column and database versions, causal length, and sequence. +- **The Node package rejects `ia32` on every operating system** rather than selecting an incompatible binary. `x64` and `arm64-musl` selection is unchanged. + ## [1.1.3] - 2026-09-11 ### Added diff --git a/Makefile b/Makefile index f9d1acf1..0860d307 100644 --- a/Makefile +++ b/Makefile @@ -93,10 +93,9 @@ TEST_TARGET = $(patsubst %.c,$(DIST_DIR)/%$(EXE), $(notdir $(TEST_SRC))) # tested directly on in-memory buffers. NT_LDFLAGS reuses the platform LDFLAGS # (which carries -lcurl) minus the shared-library-only flags (-shared on Linux, # -dynamiclib on macOS) so it links as an executable, plus the test link libs. -# -undefined dynamic_lookup is kept: the test never opens a connection, so curl's -# transport symbols are linked but never invoked. +# The deadline regression uses a loopback socket; no external service is needed. BUILD_NETTEST = build/nettest -NT_CFLAGS = $(filter-out -DCLOUDSYNC_OMIT_NETWORK,$(T_CFLAGS)) +NT_CFLAGS = $(filter-out -DCLOUDSYNC_OMIT_NETWORK,$(T_CFLAGS)) -DCLOUDSYNC_REQUEST_TIMEOUT_SECONDS=1L -DCLOUDSYNC_CONNECT_TIMEOUT_SECONDS=1L -DCLOUDSYNC_ARTIFACT_LOW_SPEED_TIME=1L -DCLOUDSYNC_ARTIFACT_TIMEOUT_SECONDS=30L NT_LDFLAGS = $(filter-out -shared -dynamiclib -headerpad_max_install_names,$(LDFLAGS)) $(T_LDFLAGS) NT_SRC = $(SRC_FILES) $(SQLITE_DIR)/sqlite3.c $(TEST_DIR)/network_unit.c NT_OBJ = $(patsubst %.c,$(BUILD_NETTEST)/%.o,$(notdir $(NT_SRC))) @@ -298,8 +297,22 @@ ifneq ($(COVERAGE),false) endif # Run only unit tests -unittest: $(TARGET) $(DIST_DIR)/unit$(EXE) +unittest: $(TARGET) $(DIST_DIR)/unit$(EXE) $(DIST_DIR)/review_regressions$(EXE) @./$(DIST_DIR)/unit$(EXE) + @./$(DIST_DIR)/review_regressions$(EXE) + +# Force pk.c's endian-conversion branch while preserving the real host ABI. +# This catches double conversion regressions even on a little-endian CI host. +$(BUILD_TEST)/pk_forced_big_endian.o: $(SRC_DIR)/pk.c $(SRC_DIR)/cloudsync_endian.h + @mkdir -p $(BUILD_TEST) + $(CC) $(T_CFLAGS) -U__BYTE_ORDER__ -D__BYTE_ORDER__=__ORDER_BIG_ENDIAN__ -c $< -o $@ + +$(DIST_DIR)/review_regressions_big_endian$(EXE): $(TEST_OBJ) $(BUILD_TEST)/pk_forced_big_endian.o + $(CC) $(filter-out $(BUILD_TEST)/pk.o $(patsubst %.c,$(BUILD_TEST)/%.o,$(notdir $(TEST_SRC))),$(TEST_OBJ)) $(BUILD_TEST)/review_regressions.o $(BUILD_TEST)/pk_forced_big_endian.o -o $@ $(T_LDFLAGS) + +.PHONY: endian-unittest +endian-unittest: $(DIST_DIR)/review_regressions_big_endian$(EXE) + @./$(DIST_DIR)/review_regressions_big_endian$(EXE) # Network-enabled unit test binary. Link it via a file rule (like dist/unit), not in # the run recipe below: on Android `make test` runs binaries on the emulator from a @@ -308,7 +321,7 @@ unittest: $(TARGET) $(DIST_DIR)/unit$(EXE) $(DIST_DIR)/network_unit$(EXE): $(CURL_LIB) $(NT_OBJ) $(CC) $(NT_OBJ) -o $@ $(NT_LDFLAGS) -# Run the network-layer unit tests (networking compiled in, no server) +# Run the network-layer unit tests (networking compiled in, loopback only) network-unittest: $(DIST_DIR)/network_unit$(EXE) @./$(DIST_DIR)/network_unit$(EXE) diff --git a/docs/internal/audit-regressions.md b/docs/internal/audit-regressions.md new file mode 100644 index 00000000..f90af4f1 --- /dev/null +++ b/docs/internal/audit-regressions.md @@ -0,0 +1,74 @@ +# Repository audit fixes + +The previously reported `MAX_PARAMS` issue is outside this change. + +## Compatibility and limits + +- PK doubles retain their deployed little-endian IEEE754 bytes. The old code + combined host conversion with manual big-endian serialization; unconditional + byte swapping now makes the historical format explicit on every architecture. + Integers keep their existing encoding. No migration is needed for supported + little-endian deployments. +- Decompressed payloads are limited to 256 MiB before allocation. Builds may + override `CLOUDSYNC_MAX_PAYLOAD_EXPANDED_SIZE`; LZ4's `INT_MAX` bound still + applies. Existing default chunk sizes are below this limit. Oversized legacy + monolithic payloads must be rechunked or used with an explicitly raised limit. +- Curl requests now have a 30-second connection deadline and a 300-second total + deadline, including reused handles. Build overrides are + `CLOUDSYNC_CONNECT_TIMEOUT_SECONDS` and `CLOUDSYNC_REQUEST_TIMEOUT_SECONDS`. +- Failed payload writes report an error and do not advance the receive cursor. + PostgreSQL's explicit RLS WITH CHECK rejection remains a skippable policy + outcome, distinct from generic SQL/permission errors; the call still reports + processed rows, but does not advance the cursor when a policy denied rows. + SQLite retains its existing per-group partial-application behavior. +- Test database files live in a private per-run temporary directory, not HOME. + The harness deletes only its own flat test files at shutdown. + +## Regression coverage + +| Area | Focused coverage | +| --- | --- | +| 64-bit clocks | Incoming column/database versions, causal length and sequence above UINT32_MAX | +| Double encoding | Golden bytes, decoding a deployed fixture, negative-value roundtrip; forced big-endian conversion build | +| Virtual-table planner | Unusable/unsupported constraints before an accepted constraint; no accepted constraints | +| Payload failures | First, middle and final PK errors; checkpoint unchanged; allocation limits | +| Metadata refill | Trigger rejects insertion of a missing column clock | +| Block LWW | Insert/update rollback on block write failure; allocation failure at each split/list/diff allocation | +| PostgreSQL ownership | Block failure while another SPI cursor is active; no invalid tuple-table cleanup | +| JSON | Root-only member lookup, string values resembling keys, nested keys, Unicode and invalid surrogates | +| Curl | Stalled loopback HTTP server, unpooled handle and pooled handle before/after reset | +| Node | ia32 rejection on all OS families and preservation of x64/arm64-musl selection | +| Fractional indexing | 4096-byte common prefix plus the existing module suite | +| Test harness | Temporary directory cleanup, memory accounting and sanitizer-safe RowID generation | + +Run `make unittest`, `make endian-unittest`, `make network-unittest`, and +`make -C modules/fractional-indexing/test run`. The network test needs permission +to bind a loopback socket; it does not contact an external service. Run Node +checks from `packages/node` with `npm test -- --run`, `npm run typecheck`, and +`npm run build`. + +PostgreSQL's `test/postgresql/full_test.sql` includes the focused audit cases in +`57_audit_regressions.sql`. Run it only against a disposable PostgreSQL instance: +the existing suite creates and drops its test databases. + +The fractional-indexing changes and its new test are inside a Git submodule; +they must be recorded there before updating the parent repository's submodule +pointer when preparing a commit. + +## Validation performed (2026-09-11) + +- macOS arm64: all 150 existing unit checks and the new audit regressions passed. +- The same SQLite suites passed AddressSanitizer and UndefinedBehaviorSanitizer + with `UBSAN_OPTIONS=halt_on_error=1`, including zero outstanding SQLite memory. +- The forced big-endian conversion regression build passed. +- All 7 network tests and all 340 fractional-indexing module tests passed. +- PostgreSQL 17, rebuilt in an isolated Linux container: 479 reported checks + passed across 57 test groups, with zero failures and no SPI cleanup warnings. +- Node: 14 tests, TypeScript checking, and CJS/ESM/declaration builds passed. +- PostgreSQL migration compatibility check and Git whitespace checks passed. + +No live cloud-service integration was run. Windows, Android and WebAssembly +runtime suites were not executed. The x86_64 runtime attempt was unavailable on +this host (`Bad CPU type in executable`, Rosetta not available). The endian +test exercises conversion logic; it is not a substitute for real big-endian +hardware testing. diff --git a/modules/fractional-indexing b/modules/fractional-indexing index b9af0ec5..ddaf4147 160000 --- a/modules/fractional-indexing +++ b/modules/fractional-indexing @@ -1 +1 @@ -Subproject commit b9af0ec5b818bca29919e1a8d42b142feb71f269 +Subproject commit ddaf4147101462b7062c549f3fad82fe9775e645 diff --git a/packages/node/src/platform.test.ts b/packages/node/src/platform.test.ts new file mode 100644 index 00000000..f5949737 --- /dev/null +++ b/packages/node/src/platform.test.ts @@ -0,0 +1,34 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { arch, platform } from 'node:os'; +import { existsSync } from 'node:fs'; +import { getCurrentPlatform, getPlatformPackageName } from './platform'; + +vi.mock('node:os', () => ({ arch: vi.fn(), platform: vi.fn() })); +vi.mock('node:fs', () => ({ existsSync: vi.fn(), readFileSync: vi.fn() })); +vi.mock('node:child_process', () => ({ execSync: vi.fn(() => 'glibc') })); + +describe('native binary architecture selection', () => { + beforeEach(() => { + vi.mocked(existsSync).mockReturnValue(false); + vi.spyOn(process.report, 'getReport').mockReturnValue({ + header: { glibcVersionRuntime: '2.38' }, + } as ReturnType); + }); + afterEach(() => vi.restoreAllMocks()); + it.each(['darwin', 'linux', 'win32'] as const)('rejects ia32 on %s', (os) => { + vi.mocked(platform).mockReturnValue(os); + vi.mocked(arch).mockReturnValue('ia32'); + expect(() => getCurrentPlatform()).toThrow(`Unsupported platform: ${os}-ia32`); + }); + it.each(['darwin', 'linux', 'win32'] as const)('keeps x64 support on %s', (os) => { + vi.mocked(platform).mockReturnValue(os); + vi.mocked(arch).mockReturnValue('x64'); + expect(getPlatformPackageName()).toBe(`@sqliteai/sqlite-sync-${os}-x86_64`); + }); + it('keeps Linux arm64 musl support', () => { + vi.mocked(platform).mockReturnValue('linux'); + vi.mocked(arch).mockReturnValue('arm64'); + vi.mocked(existsSync).mockReturnValue(true); + expect(getCurrentPlatform()).toBe('linux-arm64-musl'); + }); +}); diff --git a/packages/node/src/platform.ts b/packages/node/src/platform.ts index f40146f0..dee4fb53 100644 --- a/packages/node/src/platform.ts +++ b/packages/node/src/platform.ts @@ -97,24 +97,24 @@ export function getCurrentPlatform(): Platform { // macOS if (platformName === 'darwin') { if (archName === 'arm64') return 'darwin-arm64'; - if (archName === 'x64' || archName === 'ia32') return 'darwin-x86_64'; + if (archName === 'x64') return 'darwin-x86_64'; } // Linux (with musl detection) - if (platformName === 'linux') { + if (platformName === 'linux' && (archName === 'arm64' || archName === 'x64')) { const muslSuffix = isMusl() ? '-musl' : ''; if (archName === 'arm64') { return `linux-arm64${muslSuffix}` as Platform; } - if (archName === 'x64' || archName === 'ia32') { + if (archName === 'x64') { return `linux-x86_64${muslSuffix}` as Platform; } } // Windows if (platformName === 'win32') { - if (archName === 'x64' || archName === 'ia32') return 'win32-x86_64'; + if (archName === 'x64') return 'win32-x86_64'; } // Unsupported platform diff --git a/src/block.c b/src/block.c index dd99b266..7087b46d 100644 --- a/src/block.c +++ b/src/block.c @@ -65,7 +65,11 @@ static bool block_list_append(block_list_t *list, const char *content, size_t co block_entry_t *e = &list->entries[list->count]; e->content = cloudsync_string_ndup(content, content_len); e->position_id = position_id ? cloudsync_string_dup(position_id) : NULL; - if (!e->content) return false; + if (!e->content || (position_id && !e->position_id)) { + cloudsync_memory_free(e->content); + cloudsync_memory_free(e->position_id); + return false; + } list->count++; return true; } @@ -95,14 +99,21 @@ block_list_t *block_split(const char *text, const char *delimiter) { if (!text || !*text) { // Empty text produces a single empty block - block_list_append(list, "", 0, NULL); + if (!block_list_append(list, "", 0, NULL)) { + block_list_free(list); + return NULL; + } return list; } + if (!delimiter) delimiter = BLOCK_DEFAULT_DELIMITER; size_t dlen = strlen(delimiter); if (dlen == 0) { // No delimiter: entire text is one block - block_list_append(list, text, strlen(text), NULL); + if (!block_list_append(list, text, strlen(text), NULL)) { + block_list_free(list); + return NULL; + } return list; } @@ -176,6 +187,11 @@ static bool block_diff_append(block_diff_t *diff, block_diff_type type, const ch e->type = type; e->position_id = cloudsync_string_dup(position_id); e->content = content ? cloudsync_string_dup(content) : NULL; + if (!e->position_id || (content && !e->content)) { + cloudsync_memory_free(e->position_id); + cloudsync_memory_free(e->content); + return false; + } diff->count++; return true; } @@ -228,7 +244,7 @@ block_diff_t *block_diff(block_entry_t *old_blocks, int old_count, // Exact match — mark any skipped old blocks as REMOVED for (int si = old_scan; si < oi; si++) { if (!old_consumed[si]) { - block_diff_append(diff, BLOCK_DIFF_REMOVED, old_blocks[si].position_id, NULL); + if (!block_diff_append(diff, BLOCK_DIFF_REMOVED, old_blocks[si].position_id, NULL)) goto fail; old_consumed[si] = true; } } @@ -252,10 +268,12 @@ block_diff_t *block_diff(block_entry_t *old_blocks, int old_count, } char *new_pos = block_position_between(last_position, next_pos); - if (new_pos) { - block_diff_append(diff, BLOCK_DIFF_ADDED, new_pos, new_parts[ni]); - last_position = diff->entries[diff->count - 1].position_id; + if (!new_pos) goto fail; + { + bool appended = block_diff_append(diff, BLOCK_DIFF_ADDED, new_pos, new_parts[ni]); cloudsync_memory_free(new_pos); + if (!appended) goto fail; + last_position = diff->entries[diff->count - 1].position_id; } } } @@ -263,12 +281,16 @@ block_diff_t *block_diff(block_entry_t *old_blocks, int old_count, // Mark remaining unconsumed old blocks as REMOVED for (int oi = old_scan; oi < old_count; oi++) { if (!old_consumed[oi]) { - block_diff_append(diff, BLOCK_DIFF_REMOVED, old_blocks[oi].position_id, NULL); + if (!block_diff_append(diff, BLOCK_DIFF_REMOVED, old_blocks[oi].position_id, NULL)) goto fail; } } if (old_consumed) cloudsync_memory_free(old_consumed); return diff; +fail: + cloudsync_memory_free(old_consumed); + block_diff_free(diff); + return NULL; } // MARK: - Materialization - diff --git a/src/cloudsync.c b/src/cloudsync.c index 6cc1f01e..9a2d1724 100644 --- a/src/cloudsync.c +++ b/src/cloudsync.c @@ -192,6 +192,14 @@ struct cloudsync_context { // CLOUDSYNC_CHECKPOINT_LAST_APPLIED receive-checkpoint mode (-1 = none yet). int64_t apply_last_db_version; int64_t apply_last_seq; + + // Entries applied, and entries rejected by a row-level security policy, both + // accumulated across a receive drain so a denial in one chunk is still visible + // when a later chunk reports. Reset with cloudsync_apply_stats_reset. Kept here + // rather than derived from the apply return value, which reports payload entries + // (denied ones included) and is a tested part of the SQL surface. + int apply_rows; + int apply_denied; }; struct cloudsync_table_context { @@ -617,6 +625,25 @@ const char *cloudsync_errmsg (cloudsync_context *data) { return data->errmsg; } +void cloudsync_apply_stats_reset (cloudsync_context *data) { + if (data) { data->apply_rows = 0; data->apply_denied = 0; } +} + +// Saturating: only a receive drain resets these, so on the direct-SQL apply path they +// accumulate for the life of the connection and signed overflow would be undefined. +static void cloudsync_apply_stats_add (cloudsync_context *data, int rows, int denied) { + if (rows > 0) data->apply_rows = (data->apply_rows > INT_MAX - rows) ? INT_MAX : data->apply_rows + rows; + if (denied > 0) data->apply_denied = (data->apply_denied > INT_MAX - denied) ? INT_MAX : data->apply_denied + denied; +} + +int cloudsync_apply_rows_count (cloudsync_context *data) { + return (data) ? data->apply_rows : 0; +} + +int cloudsync_apply_denied_count (cloudsync_context *data) { + return (data) ? data->apply_denied : 0; +} + int cloudsync_errcode (cloudsync_context *data) { return data->errcode; } @@ -1325,6 +1352,7 @@ static int merge_pending_add (cloudsync_context *data, cloudsync_table_context * merge_pending_entry *e = &batch->entries[batch->count]; e->col_name = stable_col_name; e->col_value = col_value ? (dbvalue_t *)database_value_dup(col_value) : NULL; + if (col_value && !e->col_value) return DBRES_NOMEM; e->col_version = col_version; e->db_version = db_version; e->site_id_len = (site_len <= (int)sizeof(e->site_id)) ? site_len : (int)sizeof(e->site_id); @@ -1362,6 +1390,7 @@ static int merge_flush_pending (cloudsync_context *data) { int rc = DBRES_OK; bool flush_savepoint = false; + char error_message[1024] = {0}; // Nothing to write — handle sentinel-only case or skip if (batch->count == 0 && !(batch->sentinel_pending && batch->table)) { @@ -1371,7 +1400,9 @@ static int merge_flush_pending (cloudsync_context *data) { // Wrap database operations in a savepoint so that on failure (e.g. RLS // denial) the rollback properly releases all executor resources (open // relations, snapshots, plan cache) acquired during the failed statement. - flush_savepoint = (database_begin_savepoint(data, "merge_flush") == DBRES_OK); + rc = database_begin_savepoint(data, "merge_flush"); + if (rc != DBRES_OK) goto cleanup; + flush_savepoint = true; if (batch->count == 0) { // Sentinel with no winning columns (PK-only row) @@ -1456,12 +1487,15 @@ static int merge_flush_pending (cloudsync_context *data) { if (batch->cached_col_count > 0) { const char **new_names = (const char **)cloudsync_memory_realloc( batch->cached_col_names, batch->count * sizeof(const char *)); - if (new_names) { - for (int i = 0; i < batch->count; i++) { - new_names[i] = batch->entries[i].col_name; - } - batch->cached_col_names = new_names; + if (!new_names) { + batch->cached_col_count = 0; + rc = DBRES_NOMEM; + goto cleanup; + } + for (int i = 0; i < batch->count; i++) { + new_names[i] = batch->entries[i].col_name; } + batch->cached_col_names = new_names; } } @@ -1519,11 +1553,20 @@ static int merge_flush_pending (cloudsync_context *data) { } cleanup: + if (rc != DBRES_OK) snprintf(error_message, sizeof(error_message), "%s", cloudsync_errmsg(data)); merge_pending_free_entries(batch); if (flush_savepoint) { - if (rc == DBRES_OK) database_commit_savepoint(data, "merge_flush"); - else database_rollback_savepoint(data, "merge_flush"); + if (rc == DBRES_OK) { + rc = database_commit_savepoint(data, "merge_flush"); + // Snapshot here too: arriving with rc OK leaves error_message empty, and a + // commit that fails (a deadlock or serialization failure, say) sets a real + // message that the generic fallback below would otherwise replace. The + // rollback runs after, and touches the error state itself. + if (rc != DBRES_OK) snprintf(error_message, sizeof(error_message), "%s", cloudsync_errmsg(data)); + } + if (rc != DBRES_OK) database_rollback_savepoint(data, "merge_flush"); } + if (rc != DBRES_OK) cloudsync_set_error(data, error_message[0] ? error_message : "Unable to flush pending changes", rc); return rc; } @@ -1883,6 +1926,7 @@ int block_materialize_column (cloudsync_context *data, cloudsync_table_context * block_cap = new_cap; } block_values[block_count] = value ? cloudsync_string_dup(value) : cloudsync_string_dup(""); + if (!block_values[block_count]) { rc = DBRES_NOMEM; break; } block_count++; } databasevm_reset(vm); @@ -1892,7 +1936,7 @@ int block_materialize_column (cloudsync_context *data, cloudsync_table_context * // Free collected values for (int i = 0; i < block_count; i++) cloudsync_memory_free((void *)block_values[i]); if (block_values) cloudsync_memory_free((void *)block_values); - return cloudsync_set_dberror(data); + return cloudsync_set_error(data, "Unable to read block values", rc); } // Materialize text (NULL when no alive blocks) @@ -2103,7 +2147,6 @@ static int block_migrate_existing_rows (cloudsync_context *data, cloudsync_table const char *col_name = table->col_name[col_idx]; if (!col_name || !table->meta_ref || !table->blocks_ref) return DBRES_OK; - const char *delim = table->col_delimiter[col_idx] ? table->col_delimiter[col_idx] : BLOCK_DEFAULT_DELIMITER; int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET); // Phase 1: collect all existing PKs that have an alive regular col_name entry @@ -2145,9 +2188,11 @@ static int block_migrate_existing_rows (cloudsync_context *data, cloudsync_table void **new_pks = (void **)cloudsync_memory_realloc(pks, (uint64_t)(new_cap * sizeof(void *))); size_t *new_pklens = (size_t *)cloudsync_memory_realloc(pklens, (uint64_t)(new_cap * sizeof(size_t))); if (!new_pks || !new_pklens) { + for (int i = 0; i < pk_count; i++) cloudsync_memory_free((new_pks ? new_pks : pks)[i]); cloudsync_memory_free(new_pks ? new_pks : pks); cloudsync_memory_free(new_pklens ? new_pklens : pklens); databasevm_finalize(scan_vm); + cloudsync_memory_free(like_pattern); return DBRES_NOMEM; } pks = new_pks; @@ -2177,83 +2222,47 @@ static int block_migrate_existing_rows (cloudsync_context *data, cloudsync_table return DBRES_OK; } - // Phase 2: for each collected PK, read the column value, split into blocks, - // and insert into the blocks table + metadata using INSERT OR IGNORE. - - char *meta_sql = cloudsync_memory_mprintf(SQL_META_INSERT_BLOCK_IGNORE, table->meta_ref); - if (!meta_sql) { rc = DBRES_NOMEM; goto cleanup_pks; } - dbvm_t *meta_vm = NULL; - rc = databasevm_prepare(data, meta_sql, &meta_vm, 0); - cloudsync_memory_free(meta_sql); - if (rc != DBRES_OK) goto cleanup_pks; - - char *blocks_sql = cloudsync_memory_mprintf(SQL_BLOCKS_INSERT_IGNORE, table->blocks_ref); - if (!blocks_sql) { databasevm_finalize(meta_vm); rc = DBRES_NOMEM; goto cleanup_pks; } - dbvm_t *blocks_vm = NULL; - rc = databasevm_prepare(data, blocks_sql, &blocks_vm, 0); - cloudsync_memory_free(blocks_sql); - if (rc != DBRES_OK) { databasevm_finalize(meta_vm); goto cleanup_pks; } - - dbvm_t *val_vm = (dbvm_t *)table_column_lookup(table, col_name, false, NULL); - - for (int p = 0; p < pk_count; p++) { - const void *pk = pks[p]; - size_t pklen = pklens[p]; - - if (!val_vm) continue; - - // Read current column value from the base table - int bind_rc = pk_decode_prikey((char *)pk, pklen, pk_decode_bind_callback, (void *)val_vm); - if (bind_rc < 0) { databasevm_reset(val_vm); continue; } - - int step_rc = databasevm_step(val_vm); - const char *text = (step_rc == DBRES_ROW) ? database_column_text(val_vm, 0) : NULL; - // Make a copy of text before resetting val_vm, as the pointer is only valid until reset - char *text_copy = text ? cloudsync_string_dup(text) : NULL; - databasevm_reset(val_vm); - - if (!text_copy) continue; // NULL column value: nothing to migrate - - // Split text into blocks and store each one - block_list_t *blocks = block_split(text_copy, delim); - cloudsync_memory_free(text_copy); - if (!blocks) continue; - - char **positions = block_initial_positions(blocks->count); - if (positions) { - for (int b = 0; b < blocks->count; b++) { - char *block_cn = block_build_colname(col_name, positions[b]); - if (block_cn) { - // Metadata entry (skip if this block position already exists) - databasevm_bind_blob(meta_vm, 1, pk, (int)pklen); - databasevm_bind_text(meta_vm, 2, block_cn, -1); - databasevm_bind_int(meta_vm, 3, 1); // col_version = 1 (alive) - databasevm_bind_int(meta_vm, 4, db_version); - databasevm_bind_int(meta_vm, 5, cloudsync_bumpseq(data)); - databasevm_step(meta_vm); - databasevm_reset(meta_vm); - - // Block value (skip if this block position already exists) - databasevm_bind_blob(blocks_vm, 1, pk, (int)pklen); - databasevm_bind_text(blocks_vm, 2, block_cn, -1); - databasevm_bind_text(blocks_vm, 3, blocks->entries[b].content, -1); - databasevm_step(blocks_vm); - databasevm_reset(blocks_vm); - - cloudsync_memory_free(block_cn); - } - cloudsync_memory_free(positions[b]); + // Reuse the checked block writer; the scan above excludes migrated rows. + // As in local_block_insert, every failure carries a message: the migration aborts + // on one, and a bare code would surface as a blank error. Aborting is recoverable + // — the Phase 1 scan skips already-migrated rows, so a re-run resumes. + char errmsg[512]; + dbvm_t *val_vm = table_column_lookup(table, col_name, false, NULL); + if (!val_vm) { + snprintf(errmsg, sizeof(errmsg), "Missing value statement for block column \"%s\" of table \"%s\"", col_name, table->name); + rc = cloudsync_set_error(data, errmsg, DBRES_MISUSE); + } else { + rc = DBRES_OK; + } + for (int p = 0; p < pk_count && rc == DBRES_OK; p++) { + rc = pk_decode_prikey(pks[p], pklens[p], pk_decode_bind_callback, val_vm); + if (rc < 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); + rc = cloudsync_set_error(data, errmsg, DBRES_ERROR); + databasevm_reset(val_vm); + break; + } + rc = databasevm_step(val_vm); + if (rc == DBRES_ROW) { + const char *text = database_column_text(val_vm, 0); + bool has_text = text != NULL; + char *copy = text ? cloudsync_string_dup(text) : NULL; + databasevm_reset(val_vm); + if (has_text && !copy) { + snprintf(errmsg, sizeof(errmsg), "Not enough memory to migrate block column \"%s\" of table \"%s\"", col_name, table->name); + rc = cloudsync_set_error(data, errmsg, DBRES_NOMEM); + } else { + rc = DBRES_OK; } - cloudsync_memory_free(positions); + if (rc == DBRES_OK && copy) + rc = local_block_update(data, table, pks[p], pklens[p], col_idx, copy, db_version, true); + cloudsync_memory_free(copy); + } else if (rc == DBRES_DONE) { + snprintf(errmsg, sizeof(errmsg), "Unable to read block column \"%s\" of table \"%s\" while migrating: a tracked row is not visible to this connection (check the table's row-level security SELECT policy)", col_name, table->name); + rc = cloudsync_set_error(data, errmsg, DBRES_ERROR); } - block_list_free(blocks); + databasevm_reset(val_vm); } - - databasevm_finalize(meta_vm); - databasevm_finalize(blocks_vm); - rc = DBRES_OK; - -cleanup_pks: for (int i = 0; i < pk_count; i++) cloudsync_memory_free(pks[i]); cloudsync_memory_free(pks); cloudsync_memory_free(pklens); @@ -2806,6 +2815,7 @@ int cloudsync_refill_metatable (cloudsync_context *data, const char *table_name) const void *pk = (const char *)database_column_blob(vm, 0, &pklen); if (!pk) { rc = DBRES_ERROR; break; } rc = local_mark_insert_or_update_meta(table, pk, pklen, col_name, db_version, cloudsync_bumpseq(data)); + if (rc != DBRES_OK) break; } else if (rc == DBRES_DONE) { rc = DBRES_OK; break; @@ -3918,15 +3928,24 @@ static int cloudsync_payload_apply_reassembled_fragment (cloudsync_context *data rc = cloudsync_payload_apply_single_decoded_row(data, tbl, tbl_len, pk, pk_len, col_name, col_name_len, value, (size_t)total_size, col_version, db_version, site_id, site_id_len, cl, seq, pnrows); + // A denial leaves the transaction unusable until the caller's savepoint rolls it + // back, so the staged fragments cannot be dropped here. They are bounded by the + // stale-fragment cleanup instead. if (rc != DBRES_OK) goto cleanup; rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_DELETE, &vm, 0); if (rc == DBRES_OK) { databasevm_bind_text(vm, 1, value_id, -1); - int step_rc = databasevm_step(vm); - if (step_rc == DBRES_DONE) rc = DBRES_OK; + // A failed delete is deliberately tolerated rather than propagated: the value + // is already applied, so failing here would stall the cursor and re-deliver + // it, and a delete that fails once fails again on every retry. The leftover + // rows are bounded by the stale-fragment cleanup. + // (The former `if (step_rc == DBRES_DONE) rc = DBRES_OK;` only looked like a + // check: rc was already DBRES_OK from the prepare.) + databasevm_step(vm); } + cleanup: if (vm) databasevm_finalize(vm); if (value) cloudsync_memory_free(value); @@ -4099,6 +4118,10 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b // check if payload is compressed char *clone = NULL; if (header.expanded_size != 0) { + // Bound untrusted allocation sizes before passing them to LZ4's int API. + if (header.expanded_size > CLOUDSYNC_MAX_PAYLOAD_EXPANDED_SIZE || header.expanded_size > INT_MAX) { + return cloudsync_set_error(data, "Error on cloudsync_payload_apply: expanded payload exceeds limit", DBRES_MISUSE); + } clone = (char *)cloudsync_memory_alloc(header.expanded_size); if (!clone) return cloudsync_set_error(data, "Unable to allocate memory to uncompress payload", DBRES_NOMEM); @@ -4130,12 +4153,19 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b } int n = 0; rc = cloudsync_payload_apply_fragment_row(data, &row, &n); + // A denial is NOT skipped here, unlike the row path. Continuing past one + // leaves PostgreSQL's transaction unusable — the next statement fails with + // "buffer pin is not owned by resource owner" — and neither a savepoint + // around this call nor dropping the staged-fragment delete recovers it. + // A denied oversize value is therefore still a hard receive error that + // stalls the cursor. See test 58_v3_denied_checkpoint.sql. if (rc != DBRES_OK) break; applied_rows += n; buffer += seek; buf_len -= seek; } if (clone) cloudsync_memory_free(clone); + cloudsync_apply_stats_add(data, (rc == DBRES_OK) ? applied_rows : 0, 0); if (pnrows) *pnrows = applied_rows; // Advance the receive cursor only after the whole payload is applied, // gated on the caller-supplied checkpoint (a non-final chunk passes @@ -4156,6 +4186,9 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b uint16_t ncols = header.ncols; uint32_t nrows = header.nrows; int64_t last_payload_db_version = -1; + int first_error = DBRES_OK; + int denied_entries = 0; + char first_error_message[1024] = {0}; cloudsync_pk_decode_bind_context decoded_context = {.vm = vm}; // Initialize deferred column-batch merge @@ -4192,10 +4225,12 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b // Flush pending batch before any boundary change if (pk_changed || tbl_changed || db_version_changed) { + int pending_entries = batch.count; int flush_rc = merge_flush_pending(data); - if (flush_rc != DBRES_OK) { - rc = flush_rc; - // continue processing remaining rows + if (flush_rc == DBRES_POLICY_DENIED) denied_entries += (pending_entries > 0) ? pending_entries : 1; + else if (flush_rc != DBRES_OK && first_error == DBRES_OK) { + first_error = flush_rc; + snprintf(first_error_message, sizeof(first_error_message), "%s", cloudsync_errmsg(data)); } } @@ -4239,7 +4274,12 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b last_tbl_len = decoded_context.tbl_len; rc = databasevm_step(vm); + if (rc == DBRES_POLICY_DENIED) { denied_entries++; rc = DBRES_DONE; } if (rc != DBRES_DONE) { + if (first_error == DBRES_OK) { + first_error = rc; + snprintf(first_error_message, sizeof(first_error_message), "%s", cloudsync_errmsg(data)); + } // don't "break;", the error can be due to a RLS policy. // in case of error we try to apply the following changes } @@ -4251,8 +4291,13 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b // Final flush after loop { + int pending_entries = batch.count; int flush_rc = merge_flush_pending(data); - if (flush_rc != DBRES_OK && rc == DBRES_OK) rc = flush_rc; + if (flush_rc == DBRES_POLICY_DENIED) denied_entries += (pending_entries > 0) ? pending_entries : 1; + else if (flush_rc != DBRES_OK && first_error == DBRES_OK) { + first_error = flush_rc; + snprintf(first_error_message, sizeof(first_error_message), "%s", cloudsync_errmsg(data)); + } } data->pending_batch = NULL; @@ -4260,13 +4305,28 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b int rc1 = database_commit_savepoint(data, "cloudsync_payload_apply"); if (rc1 != DBRES_OK) rc = rc1; } + if (first_error != DBRES_OK) rc = first_error; // save last error (unused if function returns OK) if (rc != DBRES_OK && rc != DBRES_DONE) { - cloudsync_set_dberror(data); + cloudsync_set_error(data, first_error_message[0] ? first_error_message : "Unable to apply payload changes", rc); } if (rc == DBRES_DONE) rc = DBRES_OK; + + { + int applied = (rc == DBRES_OK) ? (int)nrows - denied_entries : 0; + cloudsync_apply_stats_add(data, (applied > 0) ? applied : 0, denied_entries); + } + + // A policy denial is permanent: those rows are not this site's to hold, so the + // cursor must still advance. Holding it back would re-deliver the same rows on + // every check forever, and a single denied row would stall every later change + // behind it. Denials are counted and reported instead (receive.denied), so + // discarding stays visible without being an error: a payload can be entirely + // denied and still be a correct outcome, since a single-row payload that + // belongs to another user is denied in full. + if (rc == DBRES_OK) { // Record the last applied (db_version, seq) and advance the receive cursor // once, gated on the caller-supplied checkpoint. A non-final chunk passes @@ -4299,6 +4359,125 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b return DBRES_OK; } +/* Shared by both backends: failures must abort the caller's statement. */ +int local_block_update(cloudsync_context *data, cloudsync_table_context *table, + const void *pk, size_t pklen, int column, const char *text, + int64_t version, bool initial) { + int rc = DBRES_NOMEM; + const char *col = table_colname(table, column); + block_list_t *old = block_list_create_empty(); + block_list_t *next = (text || initial) ? block_split(text ? text : "", table_col_delimiter(table, column)) : block_list_create_empty(); + block_diff_t *diff = NULL; + const char **parts = NULL; + dbvm_t *vm = NULL; + char *sql = NULL; + if (!old || !next) goto done; + if (!initial) { +#ifdef CLOUDSYNC_POSTGRESQL_BUILD + sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=$1 ORDER BY col_name COLLATE \"C\"", table_blocks_ref(table)); +#else + sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=?1 ORDER BY col_name COLLATE BINARY", table_blocks_ref(table)); +#endif + if (!sql) goto done; + rc = databasevm_prepare(data, sql, &vm, 0); + if (rc != DBRES_OK) goto done; + rc = databasevm_bind_blob(vm, 1, pk, (int)pklen); + if (rc != DBRES_OK) goto done; + while ((rc = databasevm_step(vm)) == DBRES_ROW) { + const char *name = database_column_text(vm, 0); + const char *value = database_column_text(vm, 1); + const char *pos = block_extract_position_id(name); + /* Literal prefix comparison: SQL LIKE would mix columns containing % or _. */ + if (pos && (size_t)(pos - name - 1) == strlen(col) && memcmp(name, col, strlen(col)) == 0) { + if (!block_list_add(old, value ? value : "", pos)) { rc = DBRES_NOMEM; goto done; } + } + } + if (rc != DBRES_DONE) goto done; + databasevm_finalize(vm); + vm = NULL; + } + rc = DBRES_NOMEM; + if (next->count) { + parts = cloudsync_memory_alloc((uint64_t)next->count * sizeof(*parts)); + if (!parts) goto done; + for (int i = 0; i < next->count; i++) parts[i] = next->entries[i].content; + } + diff = block_diff(old->entries, old->count, parts, next->count); + if (!diff) goto done; + rc = DBRES_OK; + for (int i = 0; i < diff->count; i++) { + block_diff_entry_t *entry = &diff->entries[i]; + char *name = block_build_colname(col, entry->position_id); + if (!name) { rc = DBRES_NOMEM; break; } + if (entry->type == BLOCK_DIFF_REMOVED) { + rc = local_mark_delete_block_meta(table, pk, pklen, name, version, cloudsync_bumpseq(data)); + if (rc == DBRES_OK) rc = block_delete_value_external(data, table, pk, pklen, name); + } else { + rc = local_mark_insert_or_update_meta(table, pk, pklen, name, version, cloudsync_bumpseq(data)); + dbvm_t *write = table_block_value_write_stmt(table); + if (rc == DBRES_OK && !write) rc = DBRES_MISUSE; + if (rc == DBRES_OK) rc = databasevm_bind_blob(write, 1, pk, (int)pklen); + if (rc == DBRES_OK) rc = databasevm_bind_text(write, 2, name, -1); + if (rc == DBRES_OK) rc = databasevm_bind_text(write, 3, entry->content, -1); + if (rc == DBRES_OK) rc = databasevm_step(write); + if (write) databasevm_reset(write); + if (rc == DBRES_DONE) rc = DBRES_OK; + } + cloudsync_memory_free(name); + if (rc != DBRES_OK) break; + } +done: + if (vm) databasevm_finalize(vm); + cloudsync_memory_free(sql); + cloudsync_memory_free((void *)parts); + block_diff_free(diff); + block_list_free(old); + block_list_free(next); + if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to update block metadata or values", rc); + return rc; +} + +int local_block_insert(cloudsync_context *data, cloudsync_table_context *table, + const void *pk, size_t pklen, int column, int64_t version) { + const char *colname = table_colname(table, column); + dbvm_t *vm = table_column_lookup(table, colname, false, NULL); + if (!vm) return cloudsync_set_error(data, "Missing block column statement", DBRES_MISUSE); + + // Every failure below must carry a message: databasevm_step clears the error on + // entry, so a bare code reaches the caller as a blank "not an error". + char errmsg[512]; + int rc = pk_decode_prikey((char *)pk, pklen, pk_decode_bind_callback, vm); + if (rc < 0) { + databasevm_reset(vm); + snprintf(errmsg, sizeof(errmsg), "Unable to decode the primary key of a row in \"%s\" while writing block column \"%s\"", table->name, colname); + return cloudsync_set_error(data, errmsg, DBRES_ERROR); + } + + rc = databasevm_step(vm); + char *copy = NULL; + if (rc == DBRES_ROW) { + const char *text = database_column_text(vm, 0); + copy = cloudsync_string_dup(text ? text : ""); + if (copy) rc = DBRES_OK; + else { + snprintf(errmsg, sizeof(errmsg), "Not enough memory to read block column \"%s\" of table \"%s\"", colname, table->name); + rc = cloudsync_set_error(data, errmsg, DBRES_NOMEM); + } + } + else if (rc == DBRES_DONE) { + // Reading the row back is part of the block write: without its text there is + // nothing to split. A row that its own session cannot select cannot sync at + // all, so report it here rather than leave the column silently untracked. + 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); + rc = cloudsync_set_error(data, errmsg, DBRES_ERROR); + } + // End the read cursor before writes that can invoke nested triggers/SPI errors. + databasevm_reset(vm); + if (rc == DBRES_OK) rc = local_block_update(data, table, pk, pklen, column, copy, version, true); + cloudsync_memory_free(copy); + return rc; +} + // MARK: - Payload load/store - int cloudsync_payload_get (cloudsync_context *data, char **blob, int *blob_size, int *db_version, int64_t *new_db_version) { diff --git a/src/cloudsync.h b/src/cloudsync.h index 7e72f86e..b9ee1a7d 100644 --- a/src/cloudsync.h +++ b/src/cloudsync.h @@ -18,7 +18,10 @@ extern "C" { #endif -#define CLOUDSYNC_VERSION "1.1.3" +#define CLOUDSYNC_VERSION "1.1.4" +#ifndef CLOUDSYNC_MAX_PAYLOAD_EXPANDED_SIZE +#define CLOUDSYNC_MAX_PAYLOAD_EXPANDED_SIZE (256U * 1024U * 1024U) +#endif #define CLOUDSYNC_MAX_TABLENAME_LEN 512 #define CLOUDSYNC_VALUE_NOTSET -1 @@ -107,6 +110,15 @@ int cloudsync_set_dberror (cloudsync_context *data); const char *cloudsync_errmsg (cloudsync_context *data); int cloudsync_errcode (cloudsync_context *data); void cloudsync_reset_error (cloudsync_context *data); + +// Entries applied, and entries rejected by a row-level security policy. Both counts +// accumulate across a receive drain (reset once before it) so denials in an early +// chunk are still reported by the call that finishes the drain. The applied count is +// tracked here rather than derived from the apply return value, which reports the +// payload's entry count (denied ones included) as part of the SQL surface. +void cloudsync_apply_stats_reset (cloudsync_context *data); +int cloudsync_apply_rows_count (cloudsync_context *data); +int cloudsync_apply_denied_count (cloudsync_context *data); int cloudsync_commit_hook (void *ctx); void cloudsync_rollback_hook (void *ctx); void cloudsync_set_schema (cloudsync_context *data, const char *schema); @@ -207,6 +219,8 @@ int local_mark_insert_or_update_meta (cloudsync_table_context *table, const void int local_mark_delete_meta (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq); 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); int block_delete_value_external (cloudsync_context *data, cloudsync_table_context *table, const void *pk, size_t pklen, const char *block_colname); +int local_block_update(cloudsync_context *data, cloudsync_table_context *table, const void *pk, size_t pklen, int column, const char *text, int64_t version, bool initial); +int local_block_insert(cloudsync_context *data, cloudsync_table_context *table, const void *pk, size_t pklen, int column, int64_t version); int local_drop_meta (cloudsync_table_context *table, const void *pk, size_t pklen); 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); diff --git a/src/database.h b/src/database.h index 56bb2d66..9ccdc6fe 100644 --- a/src/database.h +++ b/src/database.h @@ -25,7 +25,8 @@ typedef enum { DBRES_CONSTRAINT = 19, DBRES_MISUSE = 21, DBRES_ROW = 100, - DBRES_DONE = 101 + DBRES_DONE = 101, + DBRES_POLICY_DENIED = 1001 // PostgreSQL RLS WITH CHECK denial, not a generic SQL error } DBRES; typedef enum { diff --git a/src/network/network.c b/src/network/network.c index 038adffd..30417699 100644 --- a/src/network/network.c +++ b/src/network/network.c @@ -371,19 +371,40 @@ static bool network_curl_pool_enabled(network_data *data) { return data->curl_pool_enabled > 0; } +// API calls carry small JSON, so a cap on elapsed time is the right shape for them. +// Artifact transfers are bulk and are bounded on progress instead: 256 MiB inside a +// 300s cap would demand a sustained ~875 KB/s, killing a healthy transfer on a slow +// link. Low-speed also detects a genuine stall sooner than the absolute cap does. +static void network_curl_apply_deadlines(CURL *handle, bool is_api) { + curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT, CLOUDSYNC_CONNECT_TIMEOUT_SECONDS); + curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L); + if (is_api) { + curl_easy_setopt(handle, CURLOPT_TIMEOUT, CLOUDSYNC_REQUEST_TIMEOUT_SECONDS); + return; + } + curl_easy_setopt(handle, CURLOPT_TIMEOUT, CLOUDSYNC_ARTIFACT_TIMEOUT_SECONDS); + curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT, CLOUDSYNC_ARTIFACT_LOW_SPEED_LIMIT); + curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME, CLOUDSYNC_ARTIFACT_LOW_SPEED_TIME); +} + static CURL *network_curl_for_endpoint(network_data *data, const char *endpoint, bool *pooled) { if (pooled) *pooled = false; + bool is_api = network_endpoint_is_api(data, endpoint); if (!network_curl_pool_enabled(data)) { - return curl_easy_init(); + CURL *handle = curl_easy_init(); + if (!handle) return NULL; + network_curl_apply_deadlines(handle, is_api); + return handle; } - CURL **slot = network_endpoint_is_api(data, endpoint) ? &data->api_curl : &data->artifact_curl; + CURL **slot = is_api ? &data->api_curl : &data->artifact_curl; if (!*slot) { *slot = curl_easy_init(); } else { curl_easy_reset(*slot); } if (!*slot) return NULL; + network_curl_apply_deadlines(*slot, is_api); curl_easy_setopt(*slot, CURLOPT_MAXCONNECTS, CLOUDSYNC_CURL_MAXCONNECTS); curl_easy_setopt(*slot, CURLOPT_MAXAGE_CONN, CLOUDSYNC_CURL_MAXAGE_CONN_SECONDS); @@ -392,6 +413,36 @@ static CURL *network_curl_for_endpoint(network_data *data, const char *endpoint, return *slot; } +#if defined(CLOUDSYNC_UNITTEST) && !defined(CLOUDSYNC_OMIT_CURL) +bool network_test_curl_timeout(const char *url, bool use_pool, bool as_api) { + network_data data = {0}; + data.curl_pool_enabled = use_pool ? 1 : -1; + // Classifying the url as the check endpoint selects the API deadline policy; + // leaving every endpoint NULL selects the artifact one. + if (as_api) data.check_endpoint = (char *)url; + bool ok = true; + // The second pooled call exercises curl_easy_reset as well as initialization. + for (int i = 0; i < 2; i++) { + bool pooled = false; + CURL *handle = network_curl_for_endpoint(&data, url, &pooled); + if (!handle) { ok = false; break; } + curl_easy_setopt(handle, CURLOPT_URL, url); + curl_easy_setopt(handle, CURLOPT_PROXY, ""); + CURLcode rc = curl_easy_perform(handle); + double seconds = 0; + curl_easy_getinfo(handle, CURLINFO_TOTAL_TIME, &seconds); + // curl reports a low-speed abort as CURLE_OPERATION_TIMEDOUT as well, so only + // the budget differs between the two policies. + long budget = as_api ? CLOUDSYNC_REQUEST_TIMEOUT_SECONDS : CLOUDSYNC_ARTIFACT_LOW_SPEED_TIME; + ok = ok && rc == CURLE_OPERATION_TIMEDOUT && seconds < budget + 2; + if (!pooled) curl_easy_cleanup(handle); + } + if (data.api_curl) curl_easy_cleanup(data.api_curl); + if (data.artifact_curl) curl_easy_cleanup(data.artifact_curl); + return ok; +} +#endif + static bool network_buffer_check (network_buffer *data, size_t needed) { // alloc/resize buffer if (data->bused + needed > data->balloc) { @@ -813,13 +864,6 @@ static bool jsmn_token_eq(const char *json, const jsmntok_t *tok, const char *s) strncmp(json + tok->start, s, tok->end - tok->start) == 0); } -static int jsmn_find_key(const char *json, const jsmntok_t *tokens, int ntokens, const char *key) { - for (int i = 1; i + 1 < ntokens; i++) { - if (jsmn_token_eq(json, &tokens[i], key)) return i; - } - return -1; -} - static int jsmn_token_span(const jsmntok_t *tokens, int ntokens, int index) { if (!tokens || index < 0 || index >= ntokens) return 0; int start = tokens[index].start; @@ -871,42 +915,84 @@ static jsmntok_t *json_parse_tokens_alloc(const char *json, size_t json_len, int return tokens; } +static int jsmn_find_key(const char *json, const jsmntok_t *tokens, int ntokens, const char *key) { + int value_index; + return jsmn_find_object_value(json, tokens, ntokens, 0, key, &value_index) ? value_index - 1 : -1; +} + +static int json_hex4(const char *src) { + int value = 0; + for (int i = 0; i < 4; i++) { + unsigned char c = (unsigned char)src[i]; + int digit = c >= '0' && c <= '9' ? c - '0' : + c >= 'a' && c <= 'f' ? c - 'a' + 10 : + c >= 'A' && c <= 'F' ? c - 'A' + 10 : -1; + if (digit < 0) return -1; + value = (value << 4) | digit; + } + return value; +} + static char *json_unescape_string(const char *src, int len) { - char *out = cloudsync_memory_zeroalloc(len + 1); + char *out = cloudsync_memory_zeroalloc((uint64_t)len + 1); if (!out) return NULL; - int j = 0; - for (int i = 0; i < len; ) { - if (src[i] == '\\' && i + 1 < len) { - char c = src[i + 1]; - if (c == '"' || c == '\\' || c == '/') { out[j++] = c; i += 2; } - else if (c == 'n') { out[j++] = '\n'; i += 2; } - else if (c == 'r') { out[j++] = '\r'; i += 2; } - else if (c == 't') { out[j++] = '\t'; i += 2; } - else if (c == 'b') { out[j++] = '\b'; i += 2; } - else if (c == 'f') { out[j++] = '\f'; i += 2; } - else if (c == 'u' && i + 5 < len) { - unsigned int cp = 0; - for (int k = 0; k < 4; k++) { - char h = src[i + 2 + k]; - cp <<= 4; - if (h >= '0' && h <= '9') cp |= h - '0'; - else if (h >= 'a' && h <= 'f') cp |= 10 + h - 'a'; - else if (h >= 'A' && h <= 'F') cp |= 10 + h - 'A'; + for (int i = 0; i < len;) { + unsigned char c = (unsigned char)src[i++]; + if (c != '\\') { out[j++] = (char)c; continue; } + if (i == len) goto invalid; + c = (unsigned char)src[i++]; + switch (c) { + case '"': case '\\': case '/': out[j++] = (char)c; break; + case 'n': out[j++] = '\n'; break; + case 'r': out[j++] = '\r'; break; + case 't': out[j++] = '\t'; break; + case 'b': out[j++] = '\b'; break; + case 'f': out[j++] = '\f'; break; + case 'u': { + if (len - i < 4) goto invalid; + int cp = json_hex4(src + i); + if (cp < 0) goto invalid; + i += 4; + if (cp >= 0xD800 && cp <= 0xDBFF) { + if (len - i < 6 || src[i] != '\\' || src[i + 1] != 'u') goto invalid; + int low = json_hex4(src + i + 2); + if (low < 0xDC00 || low > 0xDFFF) goto invalid; + cp = 0x10000 + ((cp - 0xD800) << 10) + low - 0xDC00; + i += 6; + } else if (cp >= 0xDC00 && cp <= 0xDFFF) goto invalid; + // Network consumers use C strings: reject embedded NUL truncation. + if (cp == 0) goto invalid; + if (cp < 0x80) out[j++] = (char)cp; + else if (cp < 0x800) { + out[j++] = (char)(0xC0 | (cp >> 6)); + out[j++] = (char)(0x80 | (cp & 0x3F)); + } else { + if (cp >= 0x10000) { + out[j++] = (char)(0xF0 | (cp >> 18)); + out[j++] = (char)(0x80 | ((cp >> 12) & 0x3F)); + } else out[j++] = (char)(0xE0 | (cp >> 12)); + out[j++] = (char)(0x80 | ((cp >> 6) & 0x3F)); + out[j++] = (char)(0x80 | (cp & 0x3F)); } - if (cp < 0x80) { out[j++] = (char)cp; } - else { out[j++] = '?'; } // non-ASCII: replace - i += 6; + break; } - else { out[j++] = src[i]; i++; } - } else { - out[j++] = src[i]; i++; + default: goto invalid; } } out[j] = '\0'; return out; +invalid: + cloudsync_memory_free(out); + return NULL; } +#ifdef CLOUDSYNC_UNITTEST +char *network_test_unescape(const char *src) { + return json_unescape_string(src, (int)strlen(src)); +} +#endif + static char *json_extract_string(const char *json, size_t json_len, const char *key) { if (!json || json_len == 0 || !key) return NULL; @@ -925,6 +1011,12 @@ static char *json_extract_string(const char *json, size_t json_len, const char * return result; } +#ifdef CLOUDSYNC_UNITTEST +char *network_test_extract_string(const char *json, const char *key) { + return json_extract_string(json, strlen(json), key); +} +#endif + static int64_t json_extract_int(const char *json, size_t json_len, const char *key, int64_t default_value) { if (!json || json_len == 0 || !key) return default_value; @@ -1324,6 +1416,20 @@ typedef struct { int64_t send_bytes; // serialized payload bytes sent this call } sync_result; +// Gateway success responses wrap the payload in {"data": {...}}; legacy servers +// and chunk objects sliced out of a chunks array are not wrapped. Key lookups are +// scoped to one object, so a caller reading a raw response body resolves the +// payload first. Frees through *owned. Mirrors the /check unwrap below. +static const char *json_response_payload(const char *json, size_t json_len, char **owned, size_t *payload_len) { + *owned = json_extract_object_raw(json, json_len, "data"); + if (*owned) { + *payload_len = strlen(*owned); + return *owned; + } + *payload_len = json_len; + return json; +} + // Returns a malloc'd raw JSON copy of failures. ("apply" or "check"), // or NULL when the field is missing or is JSON null. Caller frees with cloudsync_memory_free. static char *json_extract_failure_stage(const char *json, size_t json_len, const char *stage_key) { @@ -1579,7 +1685,12 @@ static int network_send_payload_to_apply(sqlite3_context *context, network_data return SQLITE_ERROR; } - char *s3_url = json_extract_string(upload_res.buffer, upload_res.blen, "url"); + char *upload_payload_owned = NULL; + size_t upload_payload_len = 0; + const char *upload_payload = json_response_payload(upload_res.buffer, upload_res.blen, + &upload_payload_owned, &upload_payload_len); + char *s3_url = json_extract_string(upload_payload, upload_payload_len, "url"); + cloudsync_memory_free(upload_payload_owned); if (!s3_url) { sqlite3_result_error(context, "cloudsync_network_send_changes: missing 'url' in upload response.", -1); network_result_cleanup(&upload_res); @@ -1623,24 +1734,29 @@ void network_sync_state_update_from_response(NETWORK_RESULT *res, // BACKWARD on a rollback when a later send chunk fails, and lastOptimisticVersion // becomes the durable send checkpoint — masking a decrease would advance the // checkpoint past the rolled-back changes and silently drop them. - int64_t parsed_optimistic = json_extract_int(res->buffer, res->blen, "lastOptimisticVersion", -1); + char *state_owned = NULL; + size_t state_len = 0; + const char *state_json = json_response_payload(res->buffer, res->blen, &state_owned, &state_len); + + int64_t parsed_optimistic = json_extract_int(state_json, state_len, "lastOptimisticVersion", -1); if (parsed_optimistic >= 0) *last_optimistic_version = parsed_optimistic; - int64_t parsed_confirmed = json_extract_int(res->buffer, res->blen, "lastConfirmedVersion", -1); + int64_t parsed_confirmed = json_extract_int(state_json, state_len, "lastConfirmedVersion", -1); if (parsed_confirmed >= 0) *last_confirmed_version = parsed_confirmed; - int parsed_gaps_size = json_extract_array_size(res->buffer, res->blen, "gaps"); + int parsed_gaps_size = json_extract_array_size(state_json, state_len, "gaps"); if (parsed_gaps_size >= 0) *gaps_size = parsed_gaps_size; - char *apply_failure = json_extract_failure_stage(res->buffer, res->blen, "apply"); + char *apply_failure = json_extract_failure_stage(state_json, state_len, "apply"); if (apply_failure) { if (*apply_failure_json) cloudsync_memory_free(*apply_failure_json); *apply_failure_json = apply_failure; } - char *check_failure = json_extract_failure_stage(res->buffer, res->blen, "check"); + char *check_failure = json_extract_failure_stage(state_json, state_len, "check"); if (check_failure) { if (*check_failure_json) cloudsync_memory_free(*check_failure_json); *check_failure_json = check_failure; } + cloudsync_memory_free(state_owned); #ifdef CLOUDSYNC_NETWORK_TRACE // Full endpoint response body that the sync-state fields above were parsed from. @@ -1685,7 +1801,11 @@ void cloudsync_network_has_unsent_changes (sqlite3_context *context, int argc, s int64_t last_optimistic_version = -1; if (res.code == CLOUDSYNC_NETWORK_BUFFER && res.buffer) { - last_optimistic_version = json_extract_int(res.buffer, res.blen, "lastOptimisticVersion", -1); + char *ack_owned = NULL; + size_t ack_len = 0; + const char *ack_json = json_response_payload(res.buffer, res.blen, &ack_owned, &ack_len); + last_optimistic_version = json_extract_int(ack_json, ack_len, "lastOptimisticVersion", -1); + cloudsync_memory_free(ack_owned); } else if (res.code != CLOUDSYNC_NETWORK_OK) { network_result_to_sqlite_error(context, res, "unable to retrieve current status from remote host."); network_result_cleanup(&res); @@ -2056,7 +2176,11 @@ int cloudsync_network_check_internal(sqlite3_context *context, int *pnrows, sync if (data_json) cloudsync_memory_free(data_json); // failures.check may appear in either shape; extract opportunistically. if (out) { - char *check_failure = json_extract_failure_stage(result.buffer, result.blen, "check"); + char *failure_owned = NULL; + size_t failure_len = 0; + const char *failure_json = json_response_payload(result.buffer, result.blen, &failure_owned, &failure_len); + char *check_failure = json_extract_failure_stage(failure_json, failure_len, "check"); + cloudsync_memory_free(failure_owned); if (check_failure) { if (out->check_failure_json) cloudsync_memory_free(out->check_failure_json); out->check_failure_json = check_failure; @@ -2087,6 +2211,7 @@ int cloudsync_network_check_internal(sqlite3_context *context, int *pnrows, sync // Result of a receive drain (see network_drain_changes). typedef struct { int rows; // cumulative rows applied across the drain + int denied; // payload entries rejected by a row-level security policy int chunks; // payload chunks applied this drain int64_t bytes; // serialized payload bytes received this drain bool complete; // true iff the receive stream is fully drained (nothing pending) @@ -2111,8 +2236,11 @@ static int network_drain_changes (sqlite3_context *context, sync_result *sr, int64_t drain_prev_dbv = cloudsync_dbversion(data); sr->defer_tables = true; + // Denials accumulate on the context across every chunk of this drain, so a + // denial in an early chunk is still reported by the call that finishes it. + cloudsync_apply_stats_reset(data); + int ntries = 0; // counts only "nothing ready" (202) polls - int nrows_total = 0; // cumulative rows applied across the whole drain int nchunks = 0; // payload chunks applied this call int64_t bytes_total = 0; // serialized payload bytes received this call bool complete = true; // false iff the stream is known to have more pending @@ -2136,14 +2264,13 @@ static int network_drain_changes (sqlite3_context *context, sync_result *sr, request_max_chunks = safety_remaining; } - int nrows = 0; + int nrows = 0; // required out-param; the drain total comes from the context rc = cloudsync_network_check_internal(context, &nrows, sr, &receive_err, request_max_chunks); // a receive error (network or apply) won't fix itself across retries if (rc != SQLITE_OK) { complete = false; break; } if (sr->page_delivered) { - nrows_total += nrows; // a staged (incomplete) fragment contributes 0 - bytes_total += sr->bytes_received; + bytes_total += sr->bytes_received; // a staged (incomplete) fragment applies 0 rows nchunks += sr->chunks_received; complete = !sr->more_pending; // reflects whether the stream is finished if (!sr->more_pending) break; // final batch -> drained @@ -2172,11 +2299,18 @@ static int network_drain_changes (sqlite3_context *context, sync_result *sr, } // Compute the affected-tables union once, over the whole drain window. - if (!receive_err && rc == SQLITE_OK && nrows_total > 0) { + // Report rows actually written, not payload entries: an all-denied receive would + // otherwise claim {"rows":N,"denied":N} while tables is correctly empty. The apply + // return value still counts payload entries, which is a tested part of the SQL + // surface, so the accurate count is accumulated on the context instead. + int applied_total = cloudsync_apply_rows_count(data); + + if (!receive_err && rc == SQLITE_OK && applied_total > 0) { sr->tables_json = network_get_affected_tables(db, drain_prev_dbv); } - dr->rows = nrows_total; + dr->rows = applied_total; + dr->denied = cloudsync_apply_denied_count(data); dr->chunks = nchunks; dr->bytes = bytes_total; dr->complete = complete; @@ -2229,20 +2363,20 @@ void cloudsync_network_sync (sqlite3_context *context, int wait_ms, int max_retr char *recv_part; if (escaped_err && sr.check_failure_json) { recv_part = cloudsync_memory_mprintf( - "\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\",\"lastFailure\":%s}", - nrows_total, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped_err, sr.check_failure_json); + "\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\",\"lastFailure\":%s}", + nrows_total, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped_err, sr.check_failure_json); } else if (escaped_err) { recv_part = cloudsync_memory_mprintf( - "\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\"}", - nrows_total, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped_err); + "\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\"}", + nrows_total, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped_err); } else if (sr.check_failure_json) { recv_part = cloudsync_memory_mprintf( - "\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"lastFailure\":%s}", - nrows_total, tables, dr.chunks, (long long)dr.bytes, complete_str, sr.check_failure_json); + "\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"lastFailure\":%s}", + nrows_total, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str, sr.check_failure_json); } else { recv_part = cloudsync_memory_mprintf( - "\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s}", - nrows_total, tables, dr.chunks, (long long)dr.bytes, complete_str); + "\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s}", + nrows_total, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str); } char *buf = cloudsync_memory_mprintf("{%s,%s}", send_part, recv_part); @@ -2329,17 +2463,17 @@ static void network_receive_changes_impl (sqlite3_context *context, int max_chun char *escaped = receive_err ? json_escape_string(receive_err) : NULL; char *buf; if (escaped && sr.check_failure_json) { - buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\",\"lastFailure\":%s}}", - nrows, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped, sr.check_failure_json); + buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\",\"lastFailure\":%s}}", + nrows, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped, sr.check_failure_json); } else if (escaped) { - buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\"}}", - nrows, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped); + buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"error\":\"%s\"}}", + nrows, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str, escaped); } else if (sr.check_failure_json) { - buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"lastFailure\":%s}}", - nrows, tables, dr.chunks, (long long)dr.bytes, complete_str, sr.check_failure_json); + buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s,\"lastFailure\":%s}}", + nrows, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str, sr.check_failure_json); } else { - buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s}}", - nrows, tables, dr.chunks, (long long)dr.bytes, complete_str); + buf = cloudsync_memory_mprintf("{\"receive\":{\"rows\":%d,\"denied\":%d,\"tables\":%s,\"chunks\":%d,\"bytes\":%lld,\"complete\":%s}}", + nrows, dr.denied, tables, dr.chunks, (long long)dr.bytes, complete_str); } sqlite3_result_text(context, buf, -1, cloudsync_memory_free); if (escaped) cloudsync_memory_free(escaped); diff --git a/src/network/network_private.h b/src/network/network_private.h index 21a46fd4..bd4fc92e 100644 --- a/src/network/network_private.h +++ b/src/network/network_private.h @@ -12,6 +12,25 @@ #include #define CLOUDSYNC_DEFAULT_ADDRESS "https://cloudsync.sqlite.ai" +#ifndef CLOUDSYNC_CONNECT_TIMEOUT_SECONDS +#define CLOUDSYNC_CONNECT_TIMEOUT_SECONDS 30L +#endif +#ifndef CLOUDSYNC_REQUEST_TIMEOUT_SECONDS +#define CLOUDSYNC_REQUEST_TIMEOUT_SECONDS 300L +#endif +// Artifact transfers are bulk, so they are bounded by lack of progress rather than +// by elapsed time: a large payload on a slow link would otherwise be killed +// mid-flight. The absolute value is only a backstop against a transfer that +// trickles just fast enough to stay alive, since nothing can cancel one in flight. +#ifndef CLOUDSYNC_ARTIFACT_TIMEOUT_SECONDS +#define CLOUDSYNC_ARTIFACT_TIMEOUT_SECONDS 3600L +#endif +#ifndef CLOUDSYNC_ARTIFACT_LOW_SPEED_LIMIT +#define CLOUDSYNC_ARTIFACT_LOW_SPEED_LIMIT 1024L +#endif +#ifndef CLOUDSYNC_ARTIFACT_LOW_SPEED_TIME +#define CLOUDSYNC_ARTIFACT_LOW_SPEED_TIME 60L +#endif #define CLOUDSYNC_ENDPOINT_PREFIX "v2/cloudsync/databases" #define CLOUDSYNC_ENDPOINT_UPLOAD "upload" #define CLOUDSYNC_ENDPOINT_CHECK "check" diff --git a/src/pk.c b/src/pk.c index dcc8ca67..40e88e9d 100644 --- a/src/pk.c +++ b/src/pk.c @@ -195,13 +195,13 @@ static int pk_decode_data (const uint8_t *buffer, size_t blen, size_t *bseek, si } int pk_decode_double (const uint8_t *buffer, size_t blen, size_t *bseek, double *out) { - // Doubles are encoded as IEEE754 64-bit, big-endian. - // Convert back to host order before memcpy into double. + // Historical wire format is little-endian IEEE754, unlike integer fields. + // pk_decode_uint64 already constructs a host integer from big-endian bytes. uint64_t bits_be = 0; if (!pk_decode_uint64(buffer, blen, bseek, sizeof(uint64_t), &bits_be)) return 0; - uint64_t bits = be64_to_host(bits_be); + uint64_t bits = bswap64_u64(bits_be); double value = 0.0; memcpy(&value, &bits, sizeof(bits)); *out = value; @@ -527,12 +527,13 @@ char *pk_encode (dbvalue_t **argv, int argc, char *b, bool is_prikey, size_t *bs } break; case DBTYPE_FLOAT: { - // Encode doubles as IEEE754 64-bit, big-endian + // Encode doubles as IEEE754 64-bit, little-endian. double value = database_value_double(argv[i]); if (value < 0) {value = -value; type = DATABASE_TYPE_NEGATIVE_FLOAT;} uint64_t bits; memcpy(&bits, &value, sizeof(bits)); - bits = host_to_be64(bits); + // Preserve the deployed little-endian double wire format on all hosts. + bits = bswap64_u64(bits); bseek = pk_encode_u8(buffer, bseek, (uint8_t)type); bseek = pk_encode_uint64(buffer, bseek, bits, sizeof(bits)); } diff --git a/src/postgresql/cloudsync_postgresql.c b/src/postgresql/cloudsync_postgresql.c index 3f62d2f6..a310f96c 100644 --- a/src/postgresql/cloudsync_postgresql.c +++ b/src/postgresql/cloudsync_postgresql.c @@ -2125,50 +2125,7 @@ Datum cloudsync_insert (PG_FUNCTION_ARGS) { // Process each non-primary key column for insert or update for (int i = 0; i < table_count_cols(table); i++) { if (table_col_algo(table, i) == col_algo_block) { - // Block column: read value from base table, split into blocks, store each block - dbvm_t *val_vm = table_column_lookup(table, table_colname(table, i), false, NULL); - if (!val_vm) { rc = DBRES_ERROR; break; } - - int bind_rc = pk_decode_prikey(cleanup.pk, pklen, pk_decode_bind_callback, (void *)val_vm); - if (bind_rc < 0) { databasevm_reset(val_vm); rc = DBRES_ERROR; break; } - - int step_rc = databasevm_step(val_vm); - if (step_rc == DBRES_ROW) { - const char *text = database_column_text(val_vm, 0); - const char *delim = table_col_delimiter(table, i); - const char *col = table_colname(table, i); - - block_list_t *blocks = block_split(text ? text : "", delim); - if (blocks) { - char **positions = block_initial_positions(blocks->count); - if (positions) { - for (int b = 0; b < blocks->count; b++) { - char *block_cn = block_build_colname(col, positions[b]); - if (block_cn) { - rc = local_mark_insert_or_update_meta(table, cleanup.pk, pklen, block_cn, db_version, cloudsync_bumpseq(data)); - - // Store block value in blocks table - dbvm_t *wvm = table_block_value_write_stmt(table); - if (wvm && rc == DBRES_OK) { - databasevm_bind_blob(wvm, 1, cleanup.pk, (int)pklen); - databasevm_bind_text(wvm, 2, block_cn, -1); - databasevm_bind_text(wvm, 3, blocks->entries[b].content, -1); - databasevm_step(wvm); - databasevm_reset(wvm); - } - - cloudsync_memory_free(block_cn); - } - cloudsync_memory_free(positions[b]); - if (rc != DBRES_OK) break; - } - cloudsync_memory_free(positions); - } - block_list_free(blocks); - } - } - databasevm_reset(val_vm); - if (step_rc == DBRES_ROW || step_rc == DBRES_DONE) { if (rc == DBRES_OK) continue; } + rc = local_block_insert(data, table, cleanup.pk, pklen, i, db_version); if (rc != DBRES_OK) break; } else { rc = local_mark_insert_or_update_meta(table, cleanup.pk, pklen, table_colname(table, i), db_version, cloudsync_bumpseq(data)); @@ -2476,93 +2433,8 @@ Datum cloudsync_update_finalfn (PG_FUNCTION_ARGS) { if (dbutils_value_compare((dbvalue_t *)payload->old_values[col_index], (dbvalue_t *)payload->new_values[col_index]) != 0) { if (table_col_algo(table, i) == col_algo_block) { - // Block column: diff old and new text, emit per-block metadata changes - const char *new_text = (const char *)database_value_text(payload->new_values[col_index]); - const char *delim = table_col_delimiter(table, i); - const char *col = table_colname(table, i); - - // Read existing blocks from blocks table - block_list_t *old_blocks = block_list_create_empty(); - char *like_pattern = block_build_colname(col, "%"); - if (like_pattern && old_blocks) { - char *list_sql = cloudsync_memory_mprintf( - "SELECT col_name, col_value FROM %s WHERE pk = $1 AND col_name LIKE $2 ORDER BY col_name COLLATE \"C\"", - table_blocks_ref(table)); - if (list_sql) { - dbvm_t *list_vm = NULL; - if (databasevm_prepare(data, list_sql, &list_vm, 0) == DBRES_OK) { - databasevm_bind_blob(list_vm, 1, pk, (int)pklen); - databasevm_bind_text(list_vm, 2, like_pattern, -1); - while (databasevm_step(list_vm) == DBRES_ROW) { - const char *bcn = database_column_text(list_vm, 0); - const char *bval = database_column_text(list_vm, 1); - const char *pos = block_extract_position_id(bcn); - if (pos && old_blocks) { - block_list_add(old_blocks, bval ? bval : "", pos); - } - } - databasevm_finalize(list_vm); - } - cloudsync_memory_free(list_sql); - } - } - - // Split new text into parts (NULL text = all blocks removed) - block_list_t *new_blocks = new_text ? block_split(new_text, delim) : block_list_create_empty(); - if (new_blocks && old_blocks) { - // Build array of new content strings (NULL when count is 0) - const char **new_parts = NULL; - if (new_blocks->count > 0) { - new_parts = (const char **)cloudsync_memory_alloc( - (uint64_t)(new_blocks->count * sizeof(char *))); - if (new_parts) { - for (int b = 0; b < new_blocks->count; b++) { - new_parts[b] = new_blocks->entries[b].content; - } - } - } - - if (new_parts || new_blocks->count == 0) { - block_diff_t *diff = block_diff(old_blocks->entries, old_blocks->count, - new_parts, new_blocks->count); - if (diff) { - for (int d = 0; d < diff->count; d++) { - block_diff_entry_t *de = &diff->entries[d]; - char *block_cn = block_build_colname(col, de->position_id); - if (!block_cn) continue; - - if (de->type == BLOCK_DIFF_ADDED || de->type == BLOCK_DIFF_MODIFIED) { - rc = local_mark_insert_or_update_meta(table, pk, pklen, block_cn, - db_version, cloudsync_bumpseq(data)); - // Store block value - if (rc == DBRES_OK && table_block_value_write_stmt(table)) { - dbvm_t *wvm = table_block_value_write_stmt(table); - databasevm_bind_blob(wvm, 1, pk, (int)pklen); - databasevm_bind_text(wvm, 2, block_cn, -1); - databasevm_bind_text(wvm, 3, de->content, -1); - databasevm_step(wvm); - databasevm_reset(wvm); - } - } else if (de->type == BLOCK_DIFF_REMOVED) { - // Mark block as deleted in metadata (even col_version) - rc = local_mark_delete_block_meta(table, pk, pklen, block_cn, - db_version, cloudsync_bumpseq(data)); - // Remove from blocks table - if (rc == DBRES_OK) { - block_delete_value_external(data, table, pk, pklen, block_cn); - } - } - cloudsync_memory_free(block_cn); - if (rc != DBRES_OK) break; - } - block_diff_free(diff); - } - if (new_parts) cloudsync_memory_free((void *)new_parts); - } - } - if (new_blocks) block_list_free(new_blocks); - if (old_blocks) block_list_free(old_blocks); - if (like_pattern) cloudsync_memory_free(like_pattern); + rc = local_block_update(data, table, pk, pklen, i, + (const char *)database_value_text(payload->new_values[col_index]), db_version, false); if (rc != DBRES_OK) goto cleanup; } else { rc = local_mark_insert_or_update_meta(table, pk, pklen, table_colname(table, i), db_version, cloudsync_bumpseq(data)); diff --git a/src/postgresql/database_postgresql.c b/src/postgresql/database_postgresql.c index 0f9a50b5..d81fa3bf 100644 --- a/src/postgresql/database_postgresql.c +++ b/src/postgresql/database_postgresql.c @@ -581,6 +581,7 @@ static int map_spi_result (int rc) { static void clear_fetch_batch (pg_stmt_t *stmt) { if (!stmt) return; if (stmt->last_tuptable) { + if (SPI_tuptable == stmt->last_tuptable) SPI_tuptable = NULL; SPI_freetuptable(stmt->last_tuptable); stmt->last_tuptable = NULL; } @@ -2199,6 +2200,7 @@ int databasevm_step (dbvm_t *vm) { clear_fetch_batch(stmt); SPI_cursor_fetch(stmt->portal, true, 1); + stmt->last_tuptable = SPI_tuptable; if (SPI_processed == 0) { clear_fetch_batch(stmt); @@ -2218,6 +2220,7 @@ int databasevm_step (dbvm_t *vm) { MemoryContextReset(stmt->row_mcxt); stmt->last_tuptable = SPI_tuptable; + SPI_tuptable = NULL; stmt->current_tupdesc = stmt->last_tuptable->tupdesc; stmt->current_tuple = stmt->last_tuptable->vals[0]; rc = DBRES_ROW; @@ -2242,6 +2245,7 @@ int databasevm_step (dbvm_t *vm) { // fetch first row clear_fetch_batch(stmt); SPI_cursor_fetch(stmt->portal, true, 1); + stmt->last_tuptable = SPI_tuptable; if (SPI_processed == 0) { // No rows - close portal, don't set portal_open @@ -2262,6 +2266,7 @@ int databasevm_step (dbvm_t *vm) { MemoryContextReset(stmt->row_mcxt); stmt->last_tuptable = SPI_tuptable; + SPI_tuptable = NULL; stmt->current_tupdesc = stmt->last_tuptable->tupdesc; stmt->current_tuple = stmt->last_tuptable->vals[0]; @@ -2298,7 +2303,11 @@ int databasevm_step (dbvm_t *vm) { { MemoryContextSwitchTo(oldcontext); ErrorData *edata = CopyErrorData(); - int err = cloudsync_set_error(data, edata->message, DBRES_ERROR); + // PostgreSQL uses 42501 for both missing privileges and RLS. Only the + // executor's WITH CHECK policy rejection is safe to skip during merge. + bool policy_denied = edata->sqlerrcode == ERRCODE_INSUFFICIENT_PRIVILEGE && + edata->funcname && strcmp(edata->funcname, "ExecWithCheckOptions") == 0; + int err = cloudsync_set_error(data, edata->message, policy_denied ? DBRES_POLICY_DENIED : DBRES_ERROR); FreeErrorData(edata); FlushErrorState(); @@ -2320,10 +2329,7 @@ void databasevm_finalize (dbvm_t *vm) { { clear_fetch_batch(stmt); close_portal(stmt); - if (SPI_tuptable) { - SPI_freetuptable(SPI_tuptable); - SPI_tuptable = NULL; - } + // Only free this statement's tuple table, never another active cursor's. if (stmt->plan_is_prepared && stmt->plan) { SPI_freeplan(stmt->plan); @@ -2350,11 +2356,7 @@ void databasevm_reset (dbvm_t *vm) { clear_fetch_batch(stmt); close_portal(stmt); - // Clear global SPI tuple table if any - if (SPI_tuptable) { - SPI_freetuptable(SPI_tuptable); - SPI_tuptable = NULL; - } + // Non-row results are freed by step(); cursor results belong to last_tuptable. // Reset execution state stmt->executed_nonselect = false; diff --git a/src/sqlite/cloudsync_changes_sqlite.c b/src/sqlite/cloudsync_changes_sqlite.c index 5cd8a145..1cb831f1 100644 --- a/src/sqlite/cloudsync_changes_sqlite.c +++ b/src/sqlite/cloudsync_changes_sqlite.c @@ -275,7 +275,7 @@ int cloudsync_changesvtab_best_index (sqlite3_vtab *vtab, sqlite3_index_info *id // +512 for the extra space and for the WHERE and ORDER BY literals // memory internally manager by SQLite, so I cannot use memory_alloc here - size_t slen = (count1 * (11 + 1 + 11 + 1 + 5)) + (count2 * 11 + 1 + 5) + 512; + size_t slen = ((size_t)count1 * 32) + ((size_t)count2 * 20) + 512; char *s = (char *)sqlite3_malloc64((sqlite3_uint64)slen); if (!s) return SQLITE_NOMEM; size_t sindex= 0; @@ -285,7 +285,7 @@ int cloudsync_changesvtab_best_index (sqlite3_vtab *vtab, sqlite3_index_info *id int orderconsumed = 1; // is there a WHERE clause ? - if (count1 > 0) sindex += snprintf(s+sindex, slen-sindex, "WHERE "); + int accepted = 0; // check constraints for (int i=0; i < count1; ++i) { @@ -301,7 +301,7 @@ int cloudsync_changesvtab_best_index (sqlite3_vtab *vtab, sqlite3_index_info *id if (!opname) continue; // build next constraint - if (i > 0) sindex += snprintf(s+sindex, slen-sindex, " AND "); + sindex += snprintf(s+sindex, slen-sindex, accepted++ ? " AND " : "WHERE "); // handle special case where value is not needed if ((op == SQLITE_INDEX_CONSTRAINT_ISNULL) || (op == SQLITE_INDEX_CONSTRAINT_ISNOTNULL)) { @@ -566,12 +566,12 @@ int cloudsync_changesvtab_insert (sqlite3_vtab *vtab, int argc, sqlite3_value ** int insert_pk_len = sqlite3_value_bytes(argv[1]); const char *insert_name = (sqlite3_value_type(argv[2]) == SQLITE_NULL) ? CLOUDSYNC_TOMBSTONE_VALUE : (const char *)sqlite3_value_text(argv[2]); sqlite3_value *insert_value = argv[3]; - int64_t insert_col_version = (int64_t)sqlite3_value_int(argv[4]); - int64_t insert_db_version = (int64_t)sqlite3_value_int(argv[5]); + int64_t insert_col_version = sqlite3_value_int64(argv[4]); + int64_t insert_db_version = sqlite3_value_int64(argv[5]); const char *insert_site_id = (const char *)sqlite3_value_blob(argv[6]); int insert_site_id_len = sqlite3_value_bytes(argv[6]); - int64_t insert_cl = (int64_t)sqlite3_value_int(argv[7]); - int64_t insert_seq = (int64_t)sqlite3_value_int(argv[8]); + int64_t insert_cl = sqlite3_value_int64(argv[7]); + int64_t insert_seq = sqlite3_value_int64(argv[8]); // perform different logic for each different table algorithm if (table_algo_isgos(table)) return cloudsync_changesvtab_insert_gos(vtab, 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, (int64_t *)rowid); diff --git a/src/sqlite/cloudsync_sqlite.c b/src/sqlite/cloudsync_sqlite.c index c416b015..aa31761e 100644 --- a/src/sqlite/cloudsync_sqlite.c +++ b/src/sqlite/cloudsync_sqlite.c @@ -471,50 +471,7 @@ void dbsync_insert (sqlite3_context *context, int argc, sqlite3_value **argv) { // process each non-primary key column for insert or update for (int i=0; icount); - if (positions) { - for (int b = 0; b < blocks->count; b++) { - char *block_cn = block_build_colname(col, positions[b]); - if (block_cn) { - rc = local_mark_insert_or_update_meta(table, pk, pklen, block_cn, db_version, cloudsync_bumpseq(data)); - - // Store block value in blocks table - dbvm_t *wvm = table_block_value_write_stmt(table); - if (wvm && rc == SQLITE_OK) { - databasevm_bind_blob(wvm, 1, pk, (int)pklen); - databasevm_bind_text(wvm, 2, block_cn, -1); - databasevm_bind_text(wvm, 3, blocks->entries[b].content, -1); - databasevm_step(wvm); - databasevm_reset(wvm); - } - - cloudsync_memory_free(block_cn); - } - cloudsync_memory_free(positions[b]); - if (rc != SQLITE_OK) break; - } - cloudsync_memory_free(positions); - } - block_list_free(blocks); - } - } - databasevm_reset((dbvm_t *)val_vm); - if (rc == DBRES_ROW || rc == DBRES_DONE) rc = SQLITE_OK; + rc = local_block_insert(data, table, pk, pklen, i, db_version); if (rc != SQLITE_OK) goto cleanup; } else { // Regular column: mark as inserted or updated in the metadata @@ -743,96 +700,8 @@ void dbsync_update_final (sqlite3_context *context) { if (dbutils_value_compare(payload->old_values[col_index], payload->new_values[col_index]) != 0) { if (table_col_algo(table, i) == col_algo_block) { - // Block column: diff old and new text, emit per-block metadata changes - const char *new_text = (const char *)database_value_text(payload->new_values[col_index]); - const char *delim = table_col_delimiter(table, i); - const char *col = table_colname(table, i); - - // Read existing blocks from blocks table - block_list_t *old_blocks = block_list_create_empty(); - if (table_block_list_stmt(table)) { - char *like_pattern = block_build_colname(col, "%"); - if (like_pattern) { - // Query blocks table directly for existing block names and values - char *list_sql = cloudsync_memory_mprintf( - "SELECT col_name, col_value FROM %s WHERE pk = ?1 AND col_name LIKE ?2 ORDER BY col_name", - table_blocks_ref(table)); - if (list_sql) { - dbvm_t *list_vm = NULL; - if (databasevm_prepare(data, list_sql, &list_vm, 0) == DBRES_OK) { - databasevm_bind_blob(list_vm, 1, pk, (int)pklen); - databasevm_bind_text(list_vm, 2, like_pattern, -1); - while (databasevm_step(list_vm) == DBRES_ROW) { - const char *bcn = database_column_text(list_vm, 0); - const char *bval = database_column_text(list_vm, 1); - const char *pos = block_extract_position_id(bcn); - if (pos && old_blocks) { - block_list_add(old_blocks, bval ? bval : "", pos); - } - } - databasevm_finalize(list_vm); - } - cloudsync_memory_free(list_sql); - } - cloudsync_memory_free(like_pattern); - } - } - - // Split new text into parts (NULL text = all blocks removed) - block_list_t *new_blocks = new_text ? block_split(new_text, delim) : block_list_create_empty(); - if (new_blocks && old_blocks) { - // Build array of new content strings (NULL when count is 0) - const char **new_parts = NULL; - if (new_blocks->count > 0) { - new_parts = (const char **)cloudsync_memory_alloc( - (uint64_t)(new_blocks->count * sizeof(char *))); - if (new_parts) { - for (int b = 0; b < new_blocks->count; b++) { - new_parts[b] = new_blocks->entries[b].content; - } - } - } - - if (new_parts || new_blocks->count == 0) { - block_diff_t *diff = block_diff(old_blocks->entries, old_blocks->count, - new_parts, new_blocks->count); - if (diff) { - for (int d = 0; d < diff->count; d++) { - block_diff_entry_t *de = &diff->entries[d]; - char *block_cn = block_build_colname(col, de->position_id); - if (!block_cn) continue; - - if (de->type == BLOCK_DIFF_ADDED || de->type == BLOCK_DIFF_MODIFIED) { - rc = local_mark_insert_or_update_meta(table, pk, pklen, block_cn, - db_version, cloudsync_bumpseq(data)); - // Store block value - if (rc == SQLITE_OK && table_block_value_write_stmt(table)) { - dbvm_t *wvm = table_block_value_write_stmt(table); - databasevm_bind_blob(wvm, 1, pk, (int)pklen); - databasevm_bind_text(wvm, 2, block_cn, -1); - databasevm_bind_text(wvm, 3, de->content, -1); - databasevm_step(wvm); - databasevm_reset(wvm); - } - } else if (de->type == BLOCK_DIFF_REMOVED) { - // Mark block as deleted in metadata (even col_version) - rc = local_mark_delete_block_meta(table, pk, pklen, block_cn, - db_version, cloudsync_bumpseq(data)); - // Remove from blocks table - if (rc == SQLITE_OK) { - block_delete_value_external(data, table, pk, pklen, block_cn); - } - } - cloudsync_memory_free(block_cn); - if (rc != SQLITE_OK) break; - } - block_diff_free(diff); - } - if (new_parts) cloudsync_memory_free((void *)new_parts); - } - } - if (new_blocks) block_list_free(new_blocks); - if (old_blocks) block_list_free(old_blocks); + rc = local_block_update(data, table, pk, pklen, i, + (const char *)database_value_text(payload->new_values[col_index]), db_version, false); if (rc != SQLITE_OK) goto cleanup; } else { // Regular column: mark as updated in the metadata (columns are in cid order) diff --git a/test/network_unit.c b/test/network_unit.c index f4ccb27c..d1e60fa8 100644 --- a/test/network_unit.c +++ b/test/network_unit.c @@ -5,7 +5,8 @@ // Unit tests for the network layer's pure response-handling logic. Built with // networking ENABLED (unlike dist/unit, which is -DCLOUDSYNC_OMIT_NETWORK), so it // can call the internal functions directly on crafted in-memory NETWORK_RESULT -// buffers — no server, no sockets. +// buffers. The deadline regression also uses a stalled loopback HTTP socket; +// no external server is contacted. // #include @@ -127,7 +128,122 @@ static bool test_compute_status(void) { return ok; } +extern char *network_test_unescape(const char *); +extern char *network_test_extract_string(const char *, const char *); +static bool test_json_scope(void) { + char json[] = "{\"noise\":\"lastOptimisticVersion\",\"nested\":{\"lastConfirmedVersion\":999},\"lastOptimisticVersion\":42,\"lastConfirmedVersion\":7}"; + NETWORK_RESULT r = json_buffer(json); + int64_t optimistic = -1, confirmed = -1; + int gaps = -1; + char *apply = NULL, *check_failure = NULL; + network_sync_state_update_from_response(&r, &optimistic, &confirmed, &gaps, &apply, &check_failure); + bool ok = optimistic == 42 && confirmed == 7; + cloudsync_memory_free(apply); + cloudsync_memory_free(check_failure); + char nested[] = "{\"nested\":{\"lastOptimisticVersion\":999,\"lastConfirmedVersion\":999}}"; + r = json_buffer(nested); + apply = check_failure = NULL; + network_sync_state_update_from_response(&r, &optimistic, &confirmed, &gaps, &apply, &check_failure); + ok = ok && optimistic == 42 && confirmed == 7; + cloudsync_memory_free(apply); + cloudsync_memory_free(check_failure); + return ok; +} + +// Gateway success responses wrap the payload in {"data": ...} (API.md, "Success +// envelope"); legacy servers do not. Key lookups stay scoped to one object, so +// readers of a raw response body must unwrap first. +static bool test_json_envelope(void) { + char enveloped[] = "{\"data\":{\"nested\":{\"lastOptimisticVersion\":999}," + "\"lastOptimisticVersion\":15,\"lastConfirmedVersion\":12}}"; + NETWORK_RESULT r = json_buffer(enveloped); + int64_t optimistic = -1, confirmed = -1; + int gaps = -1; + char *apply = NULL, *check_failure = NULL; + network_sync_state_update_from_response(&r, &optimistic, &confirmed, &gaps, &apply, &check_failure); + bool ok = optimistic == 15 && confirmed == 12; + cloudsync_memory_free(apply); + cloudsync_memory_free(check_failure); + + // the enveloped 202 status payload also carries gaps and failures + char full[] = "{\"data\":{\"lastOptimisticVersion\":20,\"lastConfirmedVersion\":18," + "\"gaps\":[{\"dbVersionMin\":13,\"dbVersionMax\":15}]," + "\"failures\":{\"apply\":null,\"check\":{\"code\":\"boom\",\"retryable\":false}}}}"; + r = json_buffer(full); + apply = check_failure = NULL; + network_sync_state_update_from_response(&r, &optimistic, &confirmed, &gaps, &apply, &check_failure); + ok = ok && optimistic == 20 && confirmed == 18 && gaps == 1; + ok = ok && check_failure && strstr(check_failure, "boom"); + cloudsync_memory_free(apply); + cloudsync_memory_free(check_failure); + + // a legacy un-enveloped body still parses + char legacy[] = "{\"lastOptimisticVersion\":7,\"lastConfirmedVersion\":5}"; + r = json_buffer(legacy); + apply = check_failure = NULL; + network_sync_state_update_from_response(&r, &optimistic, &confirmed, &gaps, &apply, &check_failure); + ok = ok && optimistic == 7 && confirmed == 5; + cloudsync_memory_free(apply); + cloudsync_memory_free(check_failure); + + // key lookups remain scoped to one object: an enveloped url is not visible + // to a root-scoped read, which is why raw-response readers unwrap first + char *url = network_test_extract_string("{\"data\":{\"url\":\"https://s3/a\"}}", "url"); + ok = ok && url == NULL; + cloudsync_memory_free(url); + + // an un-enveloped chunk object sliced out of chunks[] resolves directly + url = network_test_extract_string("{\"cursor\":0,\"url\":\"https://s3/b\",\"watermark\":18}", "url"); + ok = ok && url && strcmp(url, "https://s3/b") == 0; + cloudsync_memory_free(url); + return ok; +} +static bool test_unicode(void) { + char *s = network_test_unescape("caf\\u00e9 \\u20ac \\ud83d\\ude80 \\/\\n"); + bool ok = s && strcmp(s, "caf\xc3\xa9 \xe2\x82\xac \xf0\x9f\x9a\x80 /\n") == 0; + cloudsync_memory_free(s); + const char *invalid[] = {"\\ud800", "\\udc00", "\\ud800\\u0041", "\\u0000", "\\uZZZZ", "\\u123", "\\"}; + for (size_t i = 0; i < sizeof(invalid) / sizeof(*invalid); i++) { + s = network_test_unescape(invalid[i]); + ok = ok && s == NULL; + cloudsync_memory_free(s); + } + return ok; +} + +#if !defined(_WIN32) && !defined(CLOUDSYNC_OMIT_CURL) +#include +#include +#include +extern bool network_test_curl_timeout(const char *, bool, bool); +static bool test_stalled_http_timeout(void) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) return false; + struct sockaddr_in address = {0}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + bool ok = bind(fd, (struct sockaddr *)&address, sizeof(address)) == 0 && listen(fd, 8) == 0; + socklen_t len = sizeof(address); + ok = ok && getsockname(fd, (struct sockaddr *)&address, &len) == 0; + char url[80]; + snprintf(url, sizeof(url), "http://127.0.0.1:%u/", ntohs(address.sin_port)); + // A listening socket that never sends HTTP simulates a stalled server. + // An API endpoint is bounded by elapsed time; an artifact transfer by a stall. + // Both shapes must abort against a server that accepts and then sends nothing. + if (ok) ok = network_test_curl_timeout(url, false, true) && network_test_curl_timeout(url, true, true); + if (ok) ok = network_test_curl_timeout(url, false, false) && network_test_curl_timeout(url, true, false); + close(fd); + return ok; +} +#endif + int main(void) { +#if !defined(_WIN32) && !defined(CLOUDSYNC_OMIT_CURL) + check("HTTP deadlines: API elapsed cap and artifact stall cap:", test_stalled_http_timeout()); +#endif + check("JSON keys only match root object members:", test_json_scope()); + check("Gateway data envelope is unwrapped before scoped lookups:", test_json_envelope()); + check("JSON Unicode, surrogate pairs and malformed escapes:", test_unicode()); printf("\nNetwork unit tests\n"); check("optimistic/confirmed version folds latest-valid (allows rollback):", test_optimistic_version_rollback()); check("non-buffer response is a no-op:", test_non_buffer_is_noop()); diff --git a/test/postgresql/27_rls_batch_merge.sql b/test/postgresql/27_rls_batch_merge.sql index 2ab51bfd..0e03d52b 100644 --- a/test/postgresql/27_rls_batch_merge.sql +++ b/test/postgresql/27_rls_batch_merge.sql @@ -277,6 +277,8 @@ SELECT COALESCE(max(db_version), 0) AS max_dbv_5 FROM cloudsync_changes \gset -- Apply as test_rls_user with USER1 identity — should be denied (doc4 owned by USER2) \connect cloudsync_test_27_b \ir helper_psql_conn_setup.sql +-- read the receive cursor on the target, as superuser, before dropping to the RLS role +SELECT coalesce((SELECT value::BIGINT FROM cloudsync_settings WHERE key='check_dbversion'), 0) AS ckpt_before_denied \gset SET app.current_user_id = :'USER1'; SET ROLE test_rls_user; SELECT cloudsync_payload_apply(decode(:'payload_hex_5', 'hex')) AS apply_5 \gset @@ -294,6 +296,18 @@ SELECT (:apply_5::int = 3) AS apply_5_ok \gset SELECT (:fail::int + 1) AS fail \gset \endif +-- A denial is permanent, so the cursor must still move past those rows: holding it +-- back re-delivers them on every check forever, and in a chunked batch the final +-- chunk would checkpoint past them anyway, dropping them with no report. +SELECT coalesce((SELECT value::BIGINT FROM cloudsync_settings WHERE key='check_dbversion'), 0) AS ckpt_after_denied \gset +SELECT (:ckpt_after_denied::bigint > :ckpt_before_denied::bigint) AS ckpt_ok \gset +\if :ckpt_ok +\echo [PASS] (:testid) RLS auth: denied apply still advanced the receive checkpoint +\else +\echo [FAIL] (:testid) RLS auth: denied apply left the checkpoint at :ckpt_after_denied (expected > :ckpt_before_denied) +SELECT (:fail::int + 1) AS fail \gset +\endif + -- Verify doc4 does NOT exist (superuser check) SELECT COUNT(*) AS doc4_count FROM documents WHERE id = 'doc4' \gset SELECT (:doc4_count::int = 0) AS test5_ok \gset diff --git a/test/postgresql/39_concurrent_write_apply.sql b/test/postgresql/39_concurrent_write_apply.sql index d84397ed..a9c850ec 100644 --- a/test/postgresql/39_concurrent_write_apply.sql +++ b/test/postgresql/39_concurrent_write_apply.sql @@ -81,9 +81,13 @@ BEGIN; \set ON_ERROR_ROLLBACK on SET LOCAL lock_timeout = '500ms'; +-- Expected: the apply cannot take its lock and reports the failure — locally +-- disable ON_ERROR_STOP. ON_ERROR_ROLLBACK keeps the transaction usable. +\set ON_ERROR_STOP off \if :payload_upd_ok SELECT cloudsync_payload_apply(decode(substr(:'payload_upd', 3), 'hex')) AS _blocked_apply \gset \endif +\set ON_ERROR_STOP on COMMIT; \set ON_ERROR_ROLLBACK off diff --git a/test/postgresql/57_audit_regressions.sql b/test/postgresql/57_audit_regressions.sql new file mode 100644 index 00000000..93c4f3f1 --- /dev/null +++ b/test/postgresql/57_audit_regressions.sql @@ -0,0 +1,113 @@ +-- Audit: database errors must not be mistaken for skippable RLS denials. +\set ON_ERROR_STOP on +\connect postgres +DROP DATABASE IF EXISTS cloudsync_audit_source; +DROP DATABASE IF EXISTS cloudsync_audit_target; +CREATE DATABASE cloudsync_audit_source; +CREATE DATABASE cloudsync_audit_target; +\connect cloudsync_audit_source +CREATE EXTENSION cloudsync; +CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT); +SELECT cloudsync_init('t'); +INSERT INTO t VALUES ('1','a'),('2','b'),('3','c'); +SELECT encode(cloudsync_payload_encode(tbl,pk,col_name,col_value,col_version,db_version,site_id,cl,seq),'hex') AS audit_payload FROM cloudsync_changes \gset + +\connect cloudsync_audit_target +CREATE EXTENSION cloudsync; +CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT); +SELECT cloudsync_init('t'); +CREATE TEMP TABLE audit_payload(data BYTEA); +INSERT INTO audit_payload VALUES (decode(:'audit_payload','hex')); +CREATE FUNCTION deny_audit_row() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.id = current_setting('audit.denied_id') THEN + RAISE EXCEPTION 'audit write denied'; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER deny_audit BEFORE INSERT ON t FOR EACH ROW EXECUTE FUNCTION deny_audit_row(); +DO $$ +DECLARE denied INTEGER; failed BOOLEAN; +BEGIN + FOR denied IN 1..3 LOOP + PERFORM set_config('audit.denied_id', denied::TEXT, false); + failed := false; + BEGIN + PERFORM cloudsync_payload_apply(data) FROM audit_payload; + EXCEPTION WHEN OTHERS THEN + IF SQLERRM NOT LIKE '%audit write denied%' THEN RAISE; END IF; + failed := true; + END; + IF NOT failed THEN RAISE EXCEPTION 'Payload silently ignored error at row %', denied; END IF; + IF EXISTS (SELECT FROM t) THEN RAISE EXCEPTION 'Failed payload committed partial data'; END IF; + IF coalesce((SELECT value::BIGINT FROM cloudsync_settings WHERE key='check_dbversion'),0) <> 0 THEN + RAISE EXCEPTION 'Failed payload advanced its checkpoint'; + END IF; + END LOOP; +END $$; +\echo [PASS] (57-audit) first, middle and final merge errors are propagated without checkpoint advancement + +DROP TRIGGER deny_audit ON t; +SELECT cloudsync_set_column('t','value','algo','block'); +CREATE FUNCTION deny_audit_block() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN RAISE EXCEPTION 'audit block write denied'; END $$; +CREATE TRIGGER deny_block BEFORE INSERT ON t_cloudsync_blocks FOR EACH ROW EXECUTE FUNCTION deny_audit_block(); +DO $$ +DECLARE failed BOOLEAN := false; +BEGIN + BEGIN INSERT INTO t VALUES('block','new'); + EXCEPTION WHEN OTHERS THEN failed := true; END; + IF NOT failed OR EXISTS(SELECT FROM t) THEN RAISE EXCEPTION 'Block insert failed to roll back'; END IF; +END $$; +DROP TRIGGER deny_block ON t_cloudsync_blocks; +INSERT INTO t VALUES('block','old'); +CREATE TRIGGER deny_block BEFORE INSERT ON t_cloudsync_blocks FOR EACH ROW EXECUTE FUNCTION deny_audit_block(); +DO $$ +DECLARE failed BOOLEAN := false; +BEGIN + BEGIN UPDATE t SET value='new' WHERE id='block'; + EXCEPTION WHEN OTHERS THEN failed := true; END; + IF NOT failed OR (SELECT value FROM t WHERE id='block') <> 'old' THEN + RAISE EXCEPTION 'Block update failed to roll back'; + END IF; +END $$; +\echo [PASS] (57-audit) block insert/update failures roll back base rows and metadata + +DROP TRIGGER deny_block ON t_cloudsync_blocks; + +-- Reading the row back is part of the block write, so a row the session cannot +-- select is an error — but it must be a legible one. A bare code would reach the +-- caller blank, because databasevm_step clears the error text on entry. +DO $$ BEGIN + IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'audit_block_user') THEN + CREATE ROLE audit_block_user LOGIN; + END IF; +END $$; +GRANT USAGE ON SCHEMA public TO audit_block_user; +GRANT ALL ON ALL TABLES IN SCHEMA public TO audit_block_user; +GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA public TO audit_block_user; +ALTER TABLE t ENABLE ROW LEVEL SECURITY; +CREATE POLICY t_ins ON t FOR INSERT WITH CHECK (true); +CREATE POLICY t_sel ON t FOR SELECT USING (false); +DO $$ +DECLARE msg TEXT := ''; failed BOOLEAN := false; +BEGIN + SET LOCAL ROLE audit_block_user; + BEGIN INSERT INTO t VALUES('invisible','text'); + EXCEPTION WHEN OTHERS THEN failed := true; msg := SQLERRM; END; + RESET ROLE; + IF NOT failed THEN RAISE EXCEPTION 'Unreadable block row did not report an error'; END IF; + IF coalesce(btrim(msg), '') = '' THEN RAISE EXCEPTION 'Unreadable block row reported a blank error'; END IF; + IF msg NOT LIKE '%not visible to this connection%' OR msg NOT LIKE '%value%' THEN + RAISE EXCEPTION 'Unreadable block row reported an unhelpful error: %', msg; + END IF; +END $$; +DROP POLICY t_sel ON t; +DROP POLICY t_ins ON t; +ALTER TABLE t DISABLE ROW LEVEL SECURITY; +\echo [PASS] (57-audit) an unreadable block row reports which table and column, not a blank error + +\connect postgres +DROP DATABASE cloudsync_audit_source; +DROP DATABASE cloudsync_audit_target; +\set ON_ERROR_STOP off diff --git a/test/postgresql/58_v3_denied_checkpoint.sql b/test/postgresql/58_v3_denied_checkpoint.sql new file mode 100644 index 00000000..1e0d7d34 --- /dev/null +++ b/test/postgresql/58_v3_denied_checkpoint.sql @@ -0,0 +1,117 @@ +-- A denied v3 (fragmented) value is a hard error, NOT a skipped entry. +-- +-- The row path treats a row-level security denial as permanent and skippable: it is +-- counted, skipped, and the receive cursor advances past it. The v3 path cannot do +-- the same today. Continuing past a denial leaves PostgreSQL's transaction unusable, +-- and the next statement — the checkpoint write — fails with "buffer pin is not owned +-- by resource owner TopTransaction". Neither a savepoint around the per-value apply +-- nor skipping the staged-fragment delete recovers it. +-- +-- So this test pins the behaviour that actually holds: the denial surfaces as an +-- error and the cursor does not move. That is a known gap, not a desired outcome — +-- a denied oversize value is re-delivered on every drain. Closing it needs the v3 +-- apply to leave a recoverable transaction state. +-- +-- Test 27 covers the skip-and-advance guarantee for the v2 row path. + +\set testid '58-v3-denied' +\ir helper_test_init.sql + +\connect postgres +\ir helper_psql_conn_setup.sql +DROP DATABASE IF EXISTS cloudsync_test_58_src; +DROP DATABASE IF EXISTS cloudsync_test_58_dst; +CREATE DATABASE cloudsync_test_58_src; +CREATE DATABASE cloudsync_test_58_dst; + +DO $$ BEGIN + IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'v3_denied_user') THEN + CREATE ROLE v3_denied_user LOGIN; + END IF; +END $$; + +-- Source: one oversized value, forced to fragment into several v3 chunks. +\connect cloudsync_test_58_src +\ir helper_psql_conn_setup.sql +CREATE EXTENSION IF NOT EXISTS cloudsync; +CREATE TABLE frag_rls (id TEXT PRIMARY KEY NOT NULL, note TEXT DEFAULT ''); +SELECT cloudsync_init('frag_rls', 'CLS', 1) AS _init_src \gset +SELECT cloudsync_set('payload_max_chunk_size', '1'); -- clamps to the 256KB minimum +INSERT INTO frag_rls(id, note) +VALUES ('big', repeat('A', 262144) || repeat('B', 262144) || repeat('C', 131072)); + +SELECT count(*) FILTER (WHERE get_byte(payload, 4) = 3) AS v3_chunks +FROM cloudsync_payload_chunks() \gset +SELECT (:v3_chunks::int >= 2) AS fragmented_ok \gset +\if :fragmented_ok +\echo [PASS] (:testid) oversized value fragmented into :v3_chunks v3 chunks +\else +\echo [FAIL] (:testid) expected >=2 v3 fragments, got :v3_chunks (cannot exercise the v3 path) +SELECT (:fail::int + 1) AS fail \gset +\endif + +SELECT string_agg(encode(payload, 'hex'), ',' ORDER BY chunk_index) AS chunks_hex +FROM cloudsync_payload_chunks() \gset + +-- Target: readable, but every insert is rejected by a WITH CHECK policy. +\connect cloudsync_test_58_dst +\ir helper_psql_conn_setup.sql +CREATE EXTENSION IF NOT EXISTS cloudsync; +CREATE TABLE frag_rls (id TEXT PRIMARY KEY NOT NULL, note TEXT DEFAULT ''); +SELECT cloudsync_init('frag_rls', 'CLS', 1) AS _init_dst \gset + +GRANT USAGE ON SCHEMA public TO v3_denied_user; +GRANT ALL ON ALL TABLES IN SCHEMA public TO v3_denied_user; +GRANT ALL ON ALL SEQUENCES IN SCHEMA public TO v3_denied_user; +GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA public TO v3_denied_user; +ALTER TABLE frag_rls ENABLE ROW LEVEL SECURITY; +CREATE POLICY frag_sel ON frag_rls FOR SELECT USING (true); +CREATE POLICY frag_ins ON frag_rls FOR INSERT WITH CHECK (false); + +CREATE TABLE chunk_transport(ord INT, payload BYTEA); +INSERT INTO chunk_transport(ord, payload) +SELECT ord, decode(hexval, 'hex') +FROM unnest(string_to_array(:'chunks_hex', ',')) WITH ORDINALITY AS t(hexval, ord); +GRANT ALL ON chunk_transport TO v3_denied_user; + +SELECT coalesce((SELECT value::BIGINT FROM cloudsync_settings WHERE key='check_dbversion'), 0) AS ckpt_before \gset + +-- Apply every fragment as the restricted role, one top-level statement per chunk in +-- order. The non-final fragments stage; the final one reassembles and is denied. +-- ON_ERROR_STOP is disabled around it because the denial is expected to raise. +SET ROLE v3_denied_user; +\set ON_ERROR_STOP off +SELECT format('SELECT cloudsync_payload_apply(payload) FROM chunk_transport WHERE ord = %s;', ord) +FROM chunk_transport ORDER BY ord \gexec +\set ON_ERROR_STOP on +RESET ROLE; + +-- Reconnect for clean state after the expected denial. +\connect cloudsync_test_58_dst +\ir helper_psql_conn_setup.sql + +SELECT COUNT(*) AS applied_count FROM frag_rls WHERE id = 'big' \gset +SELECT (:applied_count::int = 0) AS denied_ok \gset +\if :denied_ok +\echo [PASS] (:testid) the fragmented value was denied by the WITH CHECK policy +\else +\echo [FAIL] (:testid) expected the value to be denied, found :applied_count rows +SELECT (:fail::int + 1) AS fail \gset +\endif + +-- Known gap: the cursor does not advance, so this value is re-delivered every drain. +-- Change this to expect an advance once the v3 apply leaves a recoverable state. +SELECT coalesce((SELECT value::BIGINT FROM cloudsync_settings WHERE key='check_dbversion'), 0) AS ckpt_after \gset +SELECT (:ckpt_after::bigint = :ckpt_before::bigint) AS ckpt_pinned \gset +\if :ckpt_pinned +\echo [PASS] (:testid) known gap: a denied fragmented value leaves the checkpoint pinned at :ckpt_after +\else +\echo [FAIL] (:testid) checkpoint moved to :ckpt_after — the v3 denial gap may be closed; update this test +SELECT (:fail::int + 1) AS fail \gset +\endif + +\connect postgres +\ir helper_psql_conn_setup.sql +DROP DATABASE IF EXISTS cloudsync_test_58_src; +DROP DATABASE IF EXISTS cloudsync_test_58_dst; +DROP ROLE IF EXISTS v3_denied_user; diff --git a/test/postgresql/full_test.sql b/test/postgresql/full_test.sql index fc760828..f4e9a97d 100644 --- a/test/postgresql/full_test.sql +++ b/test/postgresql/full_test.sql @@ -64,6 +64,8 @@ \ir 54_payload_chunks_fragment_state.sql \ir 55_payload_chunks_positional_resume.sql \ir 56_many_columns.sql +\ir 57_audit_regressions.sql +\ir 58_v3_denied_checkpoint.sql -- 'Test summary' \echo '\nTest summary:' diff --git a/test/review_regressions.c b/test/review_regressions.c new file mode 100644 index 00000000..af40466e --- /dev/null +++ b/test/review_regressions.c @@ -0,0 +1,183 @@ +// Focused audit regressions. No server or on-disk database required. +#include +#include +#include +#include +#include "sqlite3.h" +#include "cloudsync.h" +#include "cloudsync_sqlite.h" +#include "utils.h" +#include "pk.h" +extern int cloudsync_changesvtab_best_index(sqlite3_vtab *, sqlite3_index_info *); +static int failures; +#define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, __LINE__, #x); failures++; } } while (0) +static sqlite3 *open_db(void) { + sqlite3 *db = NULL; + CHECK(sqlite3_open(":memory:", &db) == SQLITE_OK); + CHECK(sqlite3_cloudsync_init(db, NULL, NULL) == SQLITE_OK); + return db; +} +static int sql(sqlite3 *db, const char *query) { return sqlite3_exec(db, query, NULL, NULL, NULL); } +static int close_db(sqlite3 *db) { + CHECK(sql(db, "SELECT cloudsync_terminate()") == SQLITE_OK); + return sqlite3_close(db); +} +static int64_t scalar(sqlite3 *db, const char *query) { + sqlite3_stmt *vm = NULL; + int64_t value = INT64_MIN; + CHECK(sqlite3_prepare_v2(db, query, -1, &vm, NULL) == SQLITE_OK); + if (vm && sqlite3_step(vm) == SQLITE_ROW) value = sqlite3_column_int64(vm, 0); + else CHECK(false); + sqlite3_finalize(vm); + return value; +} +static void test_clocks_and_double(void) { + sqlite3 *db = open_db(); + CHECK(sql(db, "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT); SELECT cloudsync_init('t');") == SQLITE_OK); + CHECK(sql(db, "INSERT INTO cloudsync_changes VALUES ('t', cloudsync_pk_encode('key'), 'value', 'ok', 4294967297, 4294967300, randomblob(16), 4294967297, 4294967310);") == SQLITE_OK); + CHECK(scalar(db, "SELECT col_version FROM cloudsync_changes WHERE col_name='value'") == INT64_C(4294967297)); + CHECK(scalar(db, "SELECT db_version FROM cloudsync_changes WHERE col_name='value'") >= INT64_C(4294967300)); + CHECK(scalar(db, "SELECT seq FROM cloudsync_changes WHERE col_name='value'") == INT64_C(4294967310)); + CHECK(scalar(db, "SELECT cl FROM cloudsync_changes WHERE col_name='value'") == INT64_C(4294967297)); + CHECK(scalar(db, "SELECT count(*) FROM t WHERE value='ok'") == 1); + CHECK(scalar(db, "SELECT hex(cloudsync_pk_encode(1.0))='0102000000000000F03F'") == 1); + CHECK(scalar(db, "SELECT cloudsync_pk_decode(x'0102000000000000F03F',1)=1.0") == 1); + CHECK(scalar(db, "SELECT cloudsync_pk_decode(cloudsync_pk_encode(-1.5),1)=-1.5") == 1); + CHECK(close_db(db) == SQLITE_OK); +} +static void test_best_index(void) { + struct sqlite3_index_constraint constraints[3] = { + {.iColumn=0, .op=SQLITE_INDEX_CONSTRAINT_EQ, .usable=0}, + {.iColumn=1, .op=SQLITE_INDEX_CONSTRAINT_MATCH, .usable=1}, + {.iColumn=5, .op=SQLITE_INDEX_CONSTRAINT_GT, .usable=1} + }; + struct sqlite3_index_constraint_usage usage[3] = {{0}}; + sqlite3_index_info info = {.nConstraint=3, .aConstraint=constraints, .aConstraintUsage=usage}; + CHECK(cloudsync_changesvtab_best_index(NULL, &info) == SQLITE_OK); + CHECK(strcmp(info.idxStr, "WHERE db_version > ? ORDER BY db_version, seq ASC") == 0); + CHECK(usage[2].argvIndex == 1); + sqlite3_free(info.idxStr); + info.nConstraint = 2; + CHECK(cloudsync_changesvtab_best_index(NULL, &info) == SQLITE_OK); + CHECK(strcmp(info.idxStr, " ORDER BY db_version, seq ASC") == 0); + sqlite3_free(info.idxStr); +} +static void test_payload_errors(void) { + for (int denied = 1; denied <= 3; denied++) { + sqlite3 *source = open_db(), *target = open_db(); + const char *schema = "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL,value TEXT); SELECT cloudsync_init('t');"; + CHECK(sql(source, schema) == SQLITE_OK && sql(target, schema) == SQLITE_OK); + CHECK(sql(source, "INSERT INTO t VALUES('1','a'),('2','b'),('3','c');") == SQLITE_OK); + char trigger[256]; + snprintf(trigger, sizeof(trigger), "CREATE TRIGGER deny BEFORE INSERT ON t WHEN NEW.id='%d' BEGIN SELECT RAISE(ABORT,'denied'); END", denied); + CHECK(sql(target, trigger) == SQLITE_OK); + sqlite3_stmt *read = NULL, *write = NULL; + CHECK(sqlite3_prepare_v2(source, "SELECT cloudsync_payload_encode(tbl,pk,col_name,col_value,col_version,db_version,site_id,cl,seq) FROM cloudsync_changes", -1, &read, NULL) == SQLITE_OK); + CHECK(sqlite3_step(read) == SQLITE_ROW); + CHECK(sqlite3_prepare_v2(target, "SELECT cloudsync_payload_decode(?1)", -1, &write, NULL) == SQLITE_OK); + CHECK(sqlite3_bind_value(write, 1, sqlite3_column_value(read, 0)) == SQLITE_OK); + CHECK(sqlite3_step(write) != SQLITE_ROW); + CHECK(strstr(sqlite3_errmsg(target), "denied") != NULL); + sqlite3_finalize(write); + sqlite3_finalize(read); + CHECK(scalar(target, "SELECT coalesce((SELECT value FROM cloudsync_settings WHERE key='check_dbversion'),0)") == 0); + CHECK(sqlite3_get_autocommit(target)); + CHECK(close_db(source) == SQLITE_OK); + CHECK(close_db(target) == SQLITE_OK); + } + sqlite3 *db = open_db(); + CHECK(sql(db, "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL); SELECT cloudsync_init('t');") == SQLITE_OK); + // v1 header: request a 4GB decompression without checksum/schema requirements. + CHECK(sql(db, "SELECT cloudsync_payload_decode(x'434C535901000000FFFFFFFF00090000000000000000000000000000000000000000')") != SQLITE_OK); + CHECK(strstr(sqlite3_errmsg(db), "exceeds limit") != NULL); + CHECK(sql(db, "SELECT cloudsync_payload_decode(x'434C5359010000001000000100090000000000000000000000000000000000000000')") != SQLITE_OK); + CHECK(strstr(sqlite3_errmsg(db), "exceeds limit") != NULL); + CHECK(close_db(db) == SQLITE_OK); +} +static void test_block_write_errors(void) { + for (int update = 0; update < 2; update++) { + sqlite3 *db = open_db(); + CHECK(sql(db, "CREATE TABLE docs(id TEXT PRIMARY KEY NOT NULL, body TEXT); SELECT cloudsync_init('docs'); SELECT cloudsync_set_column('docs','body','algo','block');") == SQLITE_OK); + if (update) CHECK(sql(db, "INSERT INTO docs VALUES('1','old')") == SQLITE_OK); + CHECK(sql(db, "CREATE TRIGGER deny_block BEFORE INSERT ON docs_cloudsync_blocks BEGIN SELECT RAISE(ABORT,'block write denied'); END") == SQLITE_OK); + CHECK(sql(db, update ? "UPDATE docs SET body='new' WHERE id='1'" : "INSERT INTO docs VALUES('1','new')") != SQLITE_OK); + CHECK(scalar(db, update ? "SELECT count(*) FROM docs WHERE body='old'" : "SELECT count(*) FROM docs") == update); + CHECK(close_db(db) == SQLITE_OK); + } +} +static void test_refill_error(void) { + sqlite3 *db = open_db(); + CHECK(sql(db, "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL,a TEXT,b TEXT); SELECT cloudsync_init('t'); INSERT INTO t VALUES('1','a','b'),('2','a','b'); DELETE FROM t_cloudsync WHERE col_name='a';") == SQLITE_OK); + cloudsync_context *ctx = cloudsync_context_create(db); + CHECK(ctx && cloudsync_context_init(ctx)); + CHECK(sql(db, "CREATE TRIGGER deny_meta BEFORE INSERT ON t_cloudsync WHEN NEW.col_name='a' BEGIN SELECT RAISE(ABORT,'metadata denied'); END") == SQLITE_OK); + CHECK(cloudsync_refill_metatable(ctx, "t") != DBRES_OK); + CHECK(scalar(db, "SELECT count(*) FROM t_cloudsync WHERE col_name='a'") == 0); + cloudsync_context_free(ctx); + CHECK(close_db(db) == SQLITE_OK); +} + +static sqlite3_mem_methods memory; +static int fail_after = -1; +static bool fail_alloc(void) { + if (fail_after < 0) return false; + if (fail_after == 0) return true; + fail_after--; + return false; +} +static void *fault_malloc(int size) { return fail_alloc() ? NULL : memory.xMalloc(size); } +static void *fault_realloc(void *ptr, int size) { return fail_alloc() ? NULL : memory.xRealloc(ptr, size); } +static void test_block_oom(void) { + block_init_allocator(); + for (int kind = 0; kind < 5; kind++) { + bool succeeded = false; + for (int n = 0; n < 100 && !succeeded; n++) { + sqlite3_int64 before = sqlite3_memory_used(); + fail_after = n; + if (kind < 3) { + block_list_t *list = block_split(kind == 0 ? "" : "a\nb", kind == 1 ? "" : "\n"); + fail_after = -1; + if (list) { + CHECK(list->count == (kind == 2 ? 2 : 1)); + block_list_free(list); + succeeded = true; + } + } else if (kind == 3) { + block_entry_t old[] = {{.content="old", .position_id="a0"}, {.content="kept", .position_id="a1"}, {.content="removed", .position_id="a2"}}; + const char *parts[] = {"kept", "new"}; + block_diff_t *diff = block_diff(old, 3, parts, 2); + fail_after = -1; + if (diff) { CHECK(diff->count == 3); block_diff_free(diff); succeeded = true; } + } else { + block_list_t *list = block_list_create_empty(); + if (list) { + bool added = block_list_add(list, "content", "a0"); + fail_after = -1; + CHECK(list->count == (added ? 1 : 0)); + block_list_free(list); + succeeded = added; + } else fail_after = -1; + } + CHECK(sqlite3_memory_used() == before); + } + CHECK(succeeded); + } +} +int main(void) { + CHECK(sqlite3_config(SQLITE_CONFIG_GETMALLOC, &memory) == SQLITE_OK); + sqlite3_mem_methods faults = memory; + faults.xMalloc = fault_malloc; + faults.xRealloc = fault_realloc; + CHECK(sqlite3_config(SQLITE_CONFIG_MALLOC, &faults) == SQLITE_OK); + CHECK(sqlite3_initialize() == SQLITE_OK); + test_clocks_and_double(); + test_best_index(); + test_payload_errors(); + test_block_write_errors(); + test_refill_error(); + test_block_oom(); + cloudsync_memory_finalize(); + CHECK(sqlite3_memory_used() == 0); + printf("Audit regressions: %d failures\n", failures); + return failures ? 1 : 0; +} diff --git a/test/unit.c b/test/unit.c index ca7d6fc6..97b2ea2d 100644 --- a/test/unit.c +++ b/test/unit.c @@ -18,6 +18,7 @@ #include #else #include +#include #endif #include "pk.h" @@ -30,6 +31,7 @@ extern char *OUT_OF_MEMORY_BUFFER; extern bool force_vtab_filter_abort; extern bool force_uncompressed_blob; +static char test_directory[192]; void dbvm_reset (dbvm_t *stmt); int dbvm_count (dbvm_t *stmt, const char *value, size_t len, int type); @@ -1130,7 +1132,7 @@ bool do_alter_tables (int table_mask, sqlite3 *db, int alter_version) { } if (table_mask & TEST_NOCOLS) { - const char *sql; + const char *sql = NULL; switch (alter_version) { case 1: sql = "SELECT cloudsync_begin_alter('" CUSTOMERS_NOCOLS_TABLE "'); " @@ -1168,7 +1170,7 @@ bool do_alter_tables (int table_mask, sqlite3 *db, int alter_version) { if (table_mask & TEST_NOPRIKEYS) { // TEST a table with implicit rowid primary key - const char *sql; + const char *sql = NULL; switch (alter_version) { case 1: sql = "SELECT cloudsync_begin_alter('customers_noprikey'); " @@ -1735,7 +1737,7 @@ bool do_test_rowid (int ntest, bool print_result) { // for an explanation see https://github.com/sqliteai/sqlite-sync/blob/main/docs/RowID.md int64_t db_version = random_int64_range(1, 17179869183); int64_t seq = random_int64_range(1, 1073741823); - int64_t rowid = (db_version << 30) | seq; + int64_t rowid = (int64_t)(((uint64_t)db_version << 30) | (uint64_t)seq); int64_t value1; int64_t value2; @@ -1747,7 +1749,7 @@ bool do_test_rowid (int ntest, bool print_result) { // special case that failed in an old version int64_t db_version = 14963874252; int64_t seq = 172784902; - int64_t rowid = (db_version << 30) | seq; + int64_t rowid = (int64_t)(((uint64_t)db_version << 30) | (uint64_t)seq); int64_t value1; int64_t value2; @@ -2346,11 +2348,7 @@ bool do_test_stale_table_settings(bool cleanup_databases) { char dbpath[256]; time_t timestamp = time(NULL); - #ifdef __ANDROID__ - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-stale-%ld.sqlite", ".", timestamp); - #else - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-stale-%ld.sqlite", getenv("HOME"), timestamp); - #endif + snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-stale-%ld.sqlite", test_directory, timestamp); // Phase 1: create database, table, and init cloudsync sqlite3 *db = NULL; @@ -2417,11 +2415,7 @@ bool do_test_stale_table_settings_dropped_meta(bool cleanup_databases) { char dbpath[256]; time_t timestamp = time(NULL); - #ifdef __ANDROID__ - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-stale-meta-%ld.sqlite", ".", timestamp); - #else - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-stale-meta-%ld.sqlite", getenv("HOME"), timestamp); - #endif + snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-stale-meta-%ld.sqlite", test_directory, timestamp); // Phase 1: create database, table, and init cloudsync sqlite3 *db = NULL; @@ -4130,18 +4124,64 @@ sqlite3 *do_create_database (void) { return db; } +static bool create_test_directory(void) { +#ifdef _WIN32 + char base[MAX_PATH]; + DWORD n = GetTempPathA(sizeof(base), base); + if (!n || n >= sizeof(base)) return false; + int len = snprintf(test_directory, sizeof(test_directory), "%scloudsync-%lu-%llu", base, + (unsigned long)GetCurrentProcessId(), (unsigned long long)GetTickCount64()); + return len > 0 && len < sizeof(test_directory) && CreateDirectoryA(test_directory, NULL); +#else + const char *base = getenv("TMPDIR"); + if (!base || !*base) { +#ifdef __ANDROID__ + base = "."; // Android test runners execute from /data/local/tmp. +#else + base = "/tmp"; +#endif + } + int len = snprintf(test_directory, sizeof(test_directory), "%s/cloudsync-test-XXXXXX", base); + return len > 0 && (size_t)len < sizeof(test_directory) && mkdtemp(test_directory) != NULL; +#endif +} + +static bool remove_test_directory(void) { + char path[512]; +#ifdef _WIN32 + WIN32_FIND_DATAA entry; + snprintf(path, sizeof(path), "%s\\*", test_directory); + HANDLE handle = FindFirstFileA(path, &entry); + if (handle == INVALID_HANDLE_VALUE) return false; + do { + if (strncmp(entry.cFileName, "cloudsync-test-", 15) != 0) continue; + snprintf(path, sizeof(path), "%s\\%s", test_directory, entry.cFileName); + DeleteFileA(path); + } while (FindNextFileA(handle, &entry)); + FindClose(handle); + return RemoveDirectoryA(test_directory) != 0; +#else + DIR *dir = opendir(test_directory); + if (!dir) return false; + struct dirent *entry; + while ((entry = readdir(dir)) != NULL) { + if (strncmp(entry->d_name, "cloudsync-test-", 15) != 0) continue; + snprintf(path, sizeof(path), "%s/%s", test_directory, entry->d_name); + unlink(path); + } + closedir(dir); + return rmdir(test_directory) == 0; +#endif +} + void do_build_database_path (char buf[256], int i, time_t timestamp, int ntest) { - #ifdef __ANDROID__ - snprintf(buf, 256, "%s/cloudsync-test-%ld-%d-%d.sqlite", ".", timestamp, ntest, i); - #else - snprintf(buf, 256, "%s/cloudsync-test-%ld-%d-%d.sqlite", getenv("HOME"), timestamp, ntest, i); - #endif + snprintf(buf, 256, "%s/cloudsync-test-%ld-%d-%d.sqlite", test_directory, timestamp, ntest, i); } sqlite3 *do_create_database_file_v2 (int i, time_t timestamp, int ntest) { sqlite3 *db = NULL; - // open database in home dir + // Open database in the private per-run temporary directory. char buf[256]; do_build_database_path(buf, i, timestamp, ntest); int rc = sqlite3_open(buf, &db); @@ -9241,11 +9281,7 @@ bool do_test_block_column_reload(bool cleanup_databases) { char dbpath[256]; time_t timestamp = time(NULL); - #ifdef __ANDROID__ - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-blockreload-%ld.sqlite", ".", timestamp); - #else - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-blockreload-%ld.sqlite", getenv("HOME"), timestamp); - #endif + snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-blockreload-%ld.sqlite", test_directory, timestamp); // Phase 1: create database, table, init cloudsync, mark a column as block algo. // Use a custom delimiter so both "algo" and "delimiter" rows get persisted. @@ -9346,11 +9382,7 @@ bool do_test_block_lww_existing_data(bool cleanup_databases) { char dbpath[256]; time_t timestamp = time(NULL); - #ifdef __ANDROID__ - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-blockexist-%ld.sqlite", ".", timestamp); - #else - snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-blockexist-%ld.sqlite", getenv("HOME"), timestamp); - #endif + snprintf(dbpath, sizeof(dbpath), "%s/cloudsync-test-blockexist-%ld.sqlite", test_directory, timestamp); int rc = sqlite3_open(dbpath, &db); if (rc != SQLITE_OK) return false; @@ -13389,6 +13421,10 @@ int test_report(const char *description, bool result){ } int main (int argc, const char * argv[]) { + if (!create_test_directory()) { + fprintf(stderr, "Unable to create private test directory\n"); + return 1; + } sqlite3 *db = NULL; int result = 0; bool print_result = false; @@ -13590,6 +13626,8 @@ int main (int argc, const char * argv[]) { printf("\tleaked: %" PRId64 " B\n", memory_used); result++; } + + if (cleanup_databases) result += test_report("Temporary Directory Cleanup:", remove_test_directory()); return result; }