added rust server

2026-10-29 18:05:24 +07:00
1 parent a2aa714a83
commit b32eabcd12
@@ -5,3 +5,5 @@ data
 *.db-shm
 *.db-wal
 balance-server
+target
+__pycache__
@@ -6,3 +6,5 @@ dist/
 *.db-wal
 balance-server
 balance-server.exe
+target/
+__pycache__/
@@ -8,7 +8,12 @@

 ## Быстрая карта проекта

-Проект написан на Go 1.25.0.
+Production-сервер написан на Rust 1.90+. Go 1.25.0+ сохранён как
+эталон совместимости API/SQLite и вариант отката.
+
+Основные Rust-модули: `src/main.rs`, `runtime.rs`, `config.rs`,
+`database.rs`, `auth.rs`, `server.rs`, `sync.rs`, `go_json.rs`.
+Веб-клиент из `internal/httpapi/web` встроен в Rust executable.

 Основные Go-файлы:

@@ -77,7 +82,7 @@

 ### 5. Конкурентность

-Go server должен быть безопасен при одновременных запросах от нескольких устройств.
+Сервер должен быть безопасен при одновременных запросах от нескольких устройств.

 Особое внимание:
 - генерации sequence;
@@ -97,6 +102,11 @@ HTTP API должен возвращать стабильную структур
 Перед PR:

 ```bash
+cargo fmt --check
+cargo test --locked
+cargo clippy --locked --all-targets -- -D warnings
+cargo build --locked --release
+python3 tools/compatibility.py --rust-binary target/release/balance-server
 go test ./...
 go vet ./...
 go build ./...
@@ -121,5 +131,7 @@ go build ./...
 - [ ] DB migration учтена.
 - [ ] Concurrent requests безопасны.
 - [ ] `go test ./...` проходит.
+- [ ] Rust tests, fmt, clippy и release build проходят.
+- [ ] Differential HTTP/JSON и Go ↔ Rust SQLite takeover проходят.
 - [ ] `go vet ./...` проходит.
 - [ ] Документация обновлена.
@@ -0,0 +1,28 @@
+[package]
+name = "balance-server"
+version = "1.0.0"
+edition = "2024"
+rust-version = "1.90"
+
+[dependencies]
+axum = "0.8"
+tokio = { version = "1", features = ["full"] }
+rusqlite = { version = "0.37", features = ["bundled"] }
+chrono = { version = "0.4", features = ["serde"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = { version = "1", features = ["raw_value", "arbitrary_precision"] }
+argon2 = "0.5"
+base64 = "0.22"
+hmac = "0.12"
+sha2 = "0.10"
+rand = "0.8"
+httpdate = "1"
+hyper = { version = "1", features = ["server", "http1"] }
+hyper-util = { version = "0.1", features = ["tokio", "service"] }
+tower = { version = "0.5", features = ["util"] }
+http-body = "1"
+
+[profile.release]
+lto = true
+strip = true
+codegen-units = 1
@@ -1,14 +1,17 @@
-FROM golang:1.25.12-alpine AS build
+FROM rust:1.90-bookworm AS build
 WORKDIR /src
-COPY go.mod go.sum* ./
-RUN go mod download
-COPY . .
-RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/balance-server ./cmd/balance-server
+COPY Cargo.toml Cargo.lock ./
+COPY src ./src
+COPY internal/httpapi/web ./internal/httpapi/web
+RUN cargo build --locked --release

-FROM alpine:3.22
-RUN addgroup -S balance && adduser -S -G balance balance && mkdir /data && chown balance:balance /data
+FROM debian:bookworm-slim
+# Keep the original Alpine image's numeric identity so existing volumes remain writable.
+RUN groupadd --system --gid 101 balance \
+    && useradd --system --uid 100 --gid balance --home-dir /nonexistent --no-log-init balance \
+    && mkdir /data && chown balance:balance /data
+COPY --from=build /src/target/release/balance-server /usr/local/bin/balance-server
 USER balance
-COPY --from=build /out/balance-server /usr/local/bin/balance-server
 ENV BALANCE_ADDR=:8080 BALANCE_DB_PATH=/data/balance.db
 EXPOSE 8080
 VOLUME ["/data"]
@@ -1,6 +1,6 @@
 # Balance Server

-Self-hosted synchronization, account and web server for the Balance iOS, macOS, watchOS and Android applications. It is a single Go process with an embedded SQLite database and a complete browser client, and runs on Linux, Windows and macOS.
+Self-hosted synchronization, account and web server for the Balance iOS, macOS, watchOS and Android applications. It is a single Rust process with a bundled SQLite engine and a complete browser client, and runs on Linux, Windows and macOS.

 ## Quick start with Docker

@@ -9,26 +9,28 @@ cp .env.example .env
 openssl rand -base64 48
 # Paste the generated value into BALANCE_JWT_SECRET in .env
 docker compose up -d --build
-curl http://localhost:8080/health
+curl http://localhost:8123/health
 ```

-Open `http://localhost:8080` to register or sign in to the web application. Data is kept in the `balance-data` Docker volume. Back it up while the container is stopped.
+Open `http://localhost:8123` to register or sign in to the web application. Data is kept in the `balance-data` Docker volume. Back it up while the container is stopped.

 ## Run without Docker

-Install Go 1.25 or newer, then run:
+Install Rust 1.90 or newer, then build the locked dependencies:

 ```sh
-go mod tidy
-go build -o balance-server ./cmd/balance-server
+cargo build --locked --release
 ```

+The binary is `target/release/balance-server` (`balance-server.exe` on Windows).
+SQLite is compiled into the binary; no separate SQLite service is needed.
+
 On Linux or macOS:

 ```sh
 export BALANCE_JWT_SECRET="replace-with-a-long-random-secret"
 export BALANCE_DB_PATH="./data/balance.db"
-./balance-server
+./target/release/balance-server
 ```

 On Windows PowerShell:
@@ -36,16 +38,13 @@ On Windows PowerShell:
 ```powershell
 $env:BALANCE_JWT_SECRET = "replace-with-a-long-random-secret"
 $env:BALANCE_DB_PATH = ".\data\balance.db"
-.\balance-server.exe
+.\target\release\balance-server.exe
 ```

-Cross-compile release binaries from any Go installation:
-
-```sh
-GOOS=linux GOARCH=amd64 go build -o dist/balance-server-linux-amd64 ./cmd/balance-server
-GOOS=darwin GOARCH=arm64 go build -o dist/balance-server-macos-arm64 ./cmd/balance-server
-GOOS=windows GOARCH=amd64 go build -o dist/balance-server-windows-amd64.exe ./cmd/balance-server
-```
+For another platform, build on that platform or install its Rust target and C toolchain
+for bundled SQLite. `cargo build --locked --release --target <target>` requires the
+matching linker and C compiler. The retained Go implementation is a compatibility
+reference and rollback option: `go build -o balance-server-go ./cmd/balance-server`.

 After starting the binary, open `http://localhost:8080` in a browser. The web assets are embedded into the executable, so there is no separate frontend build, Node.js process or static directory to deploy.

@@ -83,7 +82,7 @@ Use HTTPS before exposing the server to a network. The Apple applications accept

 ```caddyfile
 balance.example.com {
-    reverse_proxy 127.0.0.1:8080
+    reverse_proxy 127.0.0.1:8123
 }
 ```

@@ -110,3 +109,35 @@ The sync endpoint accepts transactions, custom categories and monthly budgets. T
 - Back up the SQLite database and its `-wal`/`-shm` files together, or stop the server before copying the database.
 - Changing `BALANCE_JWT_SECRET` invalidates access tokens. Existing refresh sessions remain valid and will receive new access tokens after refresh.
 - The in-process authentication limiter allows 30 authentication requests per IP per minute. Use a reverse proxy for stronger public rate limiting and request logging.
+
+## Compatibility and migration
+
+The Rust server retains the HTTP/JSON API, embedded web client, environment
+variables, SQLite schema, Argon2id password hashes, JWT format, refresh sessions
+and sync cursor semantics of the Go implementation. Docker Compose still maps
+host port **8123** to container port **8080** and keeps `/data` in `balance-data`.
+Direct binary execution defaults to port 8080.
+
+Before replacing a running Go server, stop it, back up its database directory
+(including any SQLite WAL/SHM files), and preserve `BALANCE_JWT_SECRET`. Start Rust
+against the same `BALANCE_DB_PATH`; do not run both writers simultaneously.
+Existing accounts, access/refresh tokens and saved device cursors remain usable.
+See [deployment and rollback](docs/DEPLOYMENT.md) for the full procedure.
+
+Verify the implementations locally with Go 1.25+, Rust and Python 3:
+
+```sh
+cargo test --locked
+cargo fmt --check
+cargo clippy --locked --all-targets -- -D warnings
+cargo build --locked
+python3 tools/compatibility.py
+go test ./...
+go vet ./...
+go build ./...
+```
+
+The Python standard-library suite builds the Go reference, starts both binaries
+on temporary loopback ports and databases, compares API responses and assets,
+and tests SQLite takeover and rollback in both directions. To use release or
+prebuilt executables, pass `--rust-binary PATH` and optionally `--go-binary PATH`.
@@ -23,9 +23,9 @@ Android ───────────┘

 ## Исходный код

-Текущая кодовая база содержит 9 Go-файлов.
+Production runtime — один Rust-процесс: `src/main.rs`, `runtime.rs`, `config.rs`, `database.rs`, `auth.rs`, `sync.rs`, `server.rs`, `go_json.rs`. SQLite bundled через rusqlite; HTTP обслуживается axum/hyper. `runtime.rs` сохраняет socket deadlines Go (header 5s, read 15s, write 30s, idle 60s) и ограничивает graceful shutdown десятью секундами. `go_json.rs` сохраняет особенности encoding/json и исходные payload bytes. Статические файлы из `internal/httpapi/web` встроены при компиляции. Go-код сохранён как эталон совместимости и вариант rollback.

-Пути:
+Пути сохранённой Go-реализации:

 ```text
 - internal/auth
@@ -1,108 +1,85 @@
 # Deployment

-## Компоненты
+## Runtime and configuration

-```text
-Internet
-   │
- HTTPS
-   │
-   ▼
-Reverse proxy / Load Balancer
-   │
-   ▼
-BalanceServer
-   │
-   ▼
-Database
-```
+The production executable is the Rust server, built with `cargo build --locked
+--release`. SQLite is bundled into the executable. The Go implementation remains
+in `cmd/balance-server` and `internal/` as a compatibility reference and rollback
+option; build it with `go build -o balance-server-go ./cmd/balance-server`.

-## Configuration
+The Dockerfile uses a Rust 1.90 Debian Bookworm builder and a Debian Bookworm slim
+runtime, running as the unprivileged `balance` user. The image embeds the existing
+web client and uses no separate frontend process. Build dependencies and SQLite
+are compiled at image build time.
+The numeric identity remains UID 100 / GID 101, matching the original Alpine Go
+image, so existing volume permissions are preserved across the switch.

-Все environment variables и flags должны задаваться через deployment environment, а не hard-code в Go.
+`docker compose up -d --build` publishes **http://localhost:8123** (host 8123,
+container 8080). `GET /health` is the liveness endpoint. Direct binary execution
+uses `:8080` by default. Place a TLS reverse proxy in front of the published port
+for remote access; pass `X-Forwarded-Proto: https` to enable Secure session cookies.

-Перед production deployment составьте список переменных, которые реально читает текущий код:
+| Variable | Default | Purpose |
+| --- | --- | --- |
+| `BALANCE_ADDR` | `:8080` | HTTP listen address |
+| `BALANCE_DB_PATH` | `./data/balance.db` | SQLite file; Docker uses `/data/balance.db` |
+| `BALANCE_JWT_SECRET` | required | At least 32 characters; retain across migration |
+| `BALANCE_ACCESS_TTL` | `15m` | Go-style duration for access tokens |
+| `BALANCE_REFRESH_TTL` | `720h` | Go-style duration for refresh tokens |
+| `BALANCE_ALLOW_REGISTRATION` | `true` | Enable creation of accounts |
+| `BALANCE_CORS_ORIGINS` | empty | Comma-separated allowed browser origins |

-```text
- docker-compose.yml
- docker-compose.yml
- docker-compose.yml
- go.mod
- README.md
- README.md
- README.md
- Dockerfile
- Dockerfile
- Dockerfile
- internal/database/database.go
- internal/database/database.go
- internal/database/database.go
- internal/config/config.go
- internal/config/config.go
- internal/config/config.go
- internal/auth/tokens.go
- internal/auth/tokens.go
- internal/auth/tokens.go
- internal/auth/password.go
-```
+Inject secrets through the deployment environment or `.env`; never put them in
+images or source control. Compose preserves the existing named `balance-data`
+volume mounted at `/data`. Keep that volume and its project name when upgrading.

-## Docker
+## Go to Rust migration

-Если репозиторий содержит Dockerfile/compose, используйте их как canonical deployment recipe.
+1. Build and validate the Rust binary/image before stopping production. Retain the
+   previous Go executable/image and exact environment for rollback.
+2. Stop the Go service and confirm no other process writes its SQLite database.
+3. Back up the entire database directory, including `balance.db` and any
+   `balance.db-wal`/`balance.db-shm` files. Protect the backup as financial data.
+   Record the existing environment securely, especially `BALANCE_JWT_SECRET`.
+4. Start Rust with the **same database path/volume and JWT secret**. The schema is
+   unchanged: startup creates missing tables and indexes without rewriting
+   existing accounts, records, change sequences or refresh-token hashes.
+5. Check `/health`, login with an existing account, refresh an existing session,
+   and sync a device with its saved cursor. Verify another device receives edits
+   and tombstones. Inspect error rates without logging credentials or tokens.

-Не включайте secrets в Docker image.
+The existing Argon2id password hashes, HS256 JWT claims, hashed refresh tokens,
+RFC3339 timestamps, entity versions and global change sequences are shared with
+Go. No export/import, cursor reset, account recreation or forced sign-out is
+required. Do not start both implementations against the same database at once.

-## Health/readiness
+## Rollback

-Если проект предоставляет health endpoint, load balancer должен использовать его для readiness.
+Stop Rust first. Start the retained Go executable/image with the same current
+database directory and secret. Rust-written accounts, password hashes, refresh
+sessions, records, tombstones and cursors remain readable by Go. Check health,
+login, refresh and sync again. The compatibility suite exercises both directions,
+including a second takeover back to the original implementation.

-Database connectivity желательно проверять отдельно от liveness.
+For database corruption or an unrelated failed migration, stop all writers before
+restoring the complete backup directory. Restoring an older backup discards
+changes after that backup and may require clients to resynchronize; prefer normal
+binary rollback against the current database when it is healthy. Never overwrite
+an active database or delete a Compose volume as part of an upgrade.

-## Database migrations
+## Validation and operations

-Migration должна выполняться контролируемо:
+Before release run `cargo test --locked`, `cargo build --locked --release`,
+`python3 tools/compatibility.py --rust-binary target/release/balance-server`, and
+the retained reference checks `go test ./...`, `go vet ./...`, `go build ./...`.
+The differential suite uses temporary databases and loopback listeners, compares
+API/errors/security headers/assets, tests browser cookies, pagination, LWW,
+tombstones, concurrent pushes, token expiration, refresh rotation, rate limiting,
+and shared-database takeover/rollback in both directions.

-```text
-backup
-  ↓
-migration
-  ↓
-health check
-  ↓
-application rollout
-```
-
-Не запускайте destructive migration автоматически без backup/rollback plan.
-
-## Backups
-
-Финансовые данные — пользовательские данные высокой ценности.
-
-Минимум:
- регулярные DB backups;
- retention policy;
- периодическая проверка restore;
- encrypted backup storage.
-
-## Observability
-
-Рекомендуется мониторить:
- HTTP 5xx;
- auth failures;
- sync failures;
- latency;
- DB latency/errors;
- active users/devices;
- size/lag of sync change log;
- disk/storage usage.
-
-## Release checklist
-
- [ ] Tests green.
- [ ] Database migrations проверены.
- [ ] Backup создан.
- [ ] Secrets injected через secure mechanism.
- [ ] HTTPS включён.
- [ ] Health checks работают.
- [ ] Sync tested with both mobile clients.
- [ ] Rollback plan готов.
+Keep regular encrypted backups with retention and restore drills. Monitor HTTP
+5xx, sync errors, authentication failures, latency, database/change-log growth
+and free disk space. The in-memory limiter allows 30 registration/login/refresh
+attempts per remote IP per minute and resets on restart; use proxy controls when
+operating a public service. Application logs must never include passwords,
+Authorization headers or tokens.
@@ -2,29 +2,36 @@

 ## Требования

-Проект использует Go 1.25.0.
+Production-сервер использует Rust 1.90+ и bundled SQLite. Go 1.25+ нужен для сохранённой эталонной реализации и differential-проверок. Python 3 используется без сторонних пакетов.

 Проверить локальную версию:

 ```bash
+rustc --version
+cargo --version
 go version
 ```

 ## Установка зависимостей

 ```bash
+cargo fetch --locked
 go mod download
 ```

 ## Сборка

 ```bash
+cargo build --locked
 go build ./...
 ```

 ## Тесты

 ```bash
+cargo test --locked
+cargo build --locked
+python3 tools/compatibility.py
 go test ./...
 ```

@@ -37,6 +44,8 @@ go test -race ./...
 ## Static checks

 ```bash
+cargo fmt --check
+cargo clippy --locked --all-targets -- -D warnings
 go vet ./...
 ```

@@ -48,24 +57,24 @@ go vet ./...

 Не коммитьте реальные credentials.

-Пример общего workflow:
+Собрать и запустить Rust-сервер:

 ```bash
-go mod download
-go test ./...
-go build ./...
-./<server-binary>
+cargo build --locked --release
+export BALANCE_JWT_SECRET="replace-with-a-long-random-secret"
+export BALANCE_DB_PATH="./data/balance.db"
+./target/release/balance-server
 ```

-Имя binary и обязательные env vars определяются текущей entrypoint/config реализацией.
+По умолчанию сервер слушает `:8080`. Файл `.env` читает Docker Compose; при
+прямом запуске переменные нужно передать процессу через окружение.

 ## Database

-Перед первым запуском:
-1. поднять требуемую БД;
-2. применить migrations, если проект их использует;
-3. задать connection settings;
-4. проверить health/startup logs.
+SQLite встроен в executable. При первом запуске сервер создаёт каталог,
+таблицы и индексы автоматически. Отдельный процесс БД не нужен. При переходе
+с Go сохраните существующий файл БД и `BALANCE_JWT_SECRET`; подробная
+процедура и откат описаны в [DEPLOYMENT.md](DEPLOYMENT.md).

 ## Production

@@ -74,11 +83,13 @@ go build ./...
 ```text
 source
  ↓
-go test ./...
+cargo fmt --check && cargo test --locked
  ↓
-go vet ./...
+cargo clippy --locked --all-targets -- -D warnings
  ↓
-go build ./...
+cargo build --locked --release
+  ↓
+python3 tools/compatibility.py --rust-binary target/release/balance-server
  ↓
 container/package
  ↓
@@ -105,3 +116,5 @@ health check
 - concurrent sync с двух устройств;
 - login/refresh после истечения access token;
 - delete → pull на другом устройстве.
+
+Production build: `cargo build --locked --release`. Прямой запуск: `BALANCE_JWT_SECRET=<secret> target/release/balance-server`, порт 8080. Docker Compose публикует порт 8123 и сохраняет прежний volume `/data`. Процедура миграции Go → Rust и rollback: [DEPLOYMENT.md](DEPLOYMENT.md).
@@ -0,0 +1,432 @@
+use argon2::{Algorithm, Argon2, Params, Version};
+use base64::{
+    Engine, alphabet,
+    engine::{
+        DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig,
+        general_purpose::{STANDARD_NO_PAD, URL_SAFE_NO_PAD},
+    },
+};
+use chrono::{DateTime, TimeDelta, Utc};
+use hmac::{Hmac, Mac};
+use rand::{RngCore, rngs::OsRng};
+use serde::{
+    Deserialize, Deserializer, Serialize,
+    de::{MapAccess, Visitor},
+};
+use sha2::{Digest, Sha256};
+
+fn random_bytes(count: usize) -> Result<Vec<u8>, String> {
+    let mut bytes = vec![0; count];
+    OsRng
+        .try_fill_bytes(&mut bytes)
+        .map_err(|_| "could not generate random bytes".to_string())?;
+    Ok(bytes)
+}
+
+pub fn random_token(count: usize) -> Result<String, String> {
+    Ok(URL_SAFE_NO_PAD.encode(random_bytes(count)?))
+}
+
+pub fn identifier() -> Result<String, String> {
+    Ok(hex(&random_bytes(16)?))
+}
+
+fn hex(bytes: &[u8]) -> String {
+    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+pub fn token_hash(token: &str) -> String {
+    hex(&Sha256::digest(token.as_bytes()))
+}
+
+// Go's Raw{Std,URL}Encoding ignores CR/LF and permits nonzero unused bits.
+// Keep signatures canonical: this decoder is only for password data/payloads.
+fn decode_go_base64(value: &str, url_safe: bool) -> Result<Vec<u8>, base64::DecodeError> {
+    let config = GeneralPurposeConfig::new()
+        .with_decode_padding_mode(DecodePaddingMode::RequireNone)
+        .with_decode_allow_trailing_bits(true);
+    let engine = GeneralPurpose::new(
+        if url_safe {
+            &alphabet::URL_SAFE
+        } else {
+            &alphabet::STANDARD
+        },
+        config,
+    );
+    let cleaned: Vec<u8> = value
+        .bytes()
+        .filter(|byte| !matches!(byte, b'\r' | b'\n'))
+        .collect();
+    engine.decode(cleaned)
+}
+
+pub fn hash_password(password: &str) -> Result<String, String> {
+    if password.len() < 8 {
+        return Err("password must contain at least 8 characters".into());
+    }
+    if password.len() > 1024 {
+        return Err("password is too long".into());
+    }
+    let salt = random_bytes(16)?;
+    let mut hash = [0; 32];
+    let params = Params::new(64 * 1024, 3, 2, Some(32)).map_err(|error| error.to_string())?;
+    Argon2::new(Algorithm::Argon2id, Version::V0x13, params)
+        .hash_password_into(password.as_bytes(), &salt, &mut hash)
+        .map_err(|error| error.to_string())?;
+    Ok(format!(
+        "$argon2id$v=19$m=65536,t=3,p=2${}${}",
+        STANDARD_NO_PAD.encode(salt),
+        STANDARD_NO_PAD.encode(hash)
+    ))
+}
+
+pub fn verify_password(encoded: &str, password: &str) -> bool {
+    let parts: Vec<_> = encoded.split('$').collect();
+    if parts.len() != 6 || parts[1] != "argon2id" || parts[2] != "v=19" {
+        return false;
+    }
+    let Some((memory, iterations, threads)) = password_parameters(parts[3]) else {
+        return false;
+    };
+    if memory > 256 * 1024 || iterations > 10 || threads > 16 {
+        return false;
+    }
+    let (Ok(salt), Ok(expected)) = (
+        decode_go_base64(parts[4], false),
+        decode_go_base64(parts[5], false),
+    ) else {
+        return false;
+    };
+    if expected.is_empty() {
+        return false;
+    }
+    let Ok(params) = Params::new(memory, iterations, threads, Some(expected.len())) else {
+        return false;
+    };
+    let mut actual = vec![0; expected.len()];
+    if Argon2::new(Algorithm::Argon2id, Version::V0x13, params)
+        .hash_password_into(password.as_bytes(), &salt, &mut actual)
+        .is_err()
+    {
+        return false;
+    }
+    // Use the MAC crate's constant-time verification primitive for equality.
+    let mut expected_mac = Hmac::<Sha256>::new_from_slice(b"balance password comparison")
+        .expect("HMAC accepts any key size");
+    expected_mac.update(&expected);
+    let expected_tag = expected_mac.finalize().into_bytes();
+    let mut actual_mac = Hmac::<Sha256>::new_from_slice(b"balance password comparison")
+        .expect("HMAC accepts any key size");
+    actual_mac.update(&actual);
+    actual_mac.verify_slice(&expected_tag).is_ok()
+}
+
+fn password_parameters(mut text: &str) -> Option<(u32, u32, u32)> {
+    fn scan(text: &mut &str, prefix: &str) -> Option<u32> {
+        *text = text.strip_prefix(prefix)?;
+        *text = text.trim_start_matches(|c: char| c.is_whitespace() && c != '\n');
+        let digits = *text;
+        let end = digits
+            .find(|c: char| !c.is_ascii_digit())
+            .unwrap_or(digits.len());
+        let value = digits[..end].parse().ok()?;
+        *text = &digits[end..];
+        Some(value)
+    }
+    let memory = scan(&mut text, "m=")?;
+    let iterations = scan(&mut text, ",t=")?;
+    let threads = scan(&mut text, ",p=")?;
+    // fmt.Sscanf stops after its final conversion and accepts a trailing suffix.
+    if threads > u8::MAX as u32 {
+        return None;
+    }
+    Some((memory, iterations, threads))
+}
+
+#[derive(Clone, Default, Debug, Serialize)]
+pub struct Claims {
+    #[serde(rename = "sub")]
+    pub subject: String,
+    pub email: String,
+    #[serde(rename = "iss")]
+    pub issuer: String,
+    #[serde(rename = "iat")]
+    pub issued: i64,
+    #[serde(rename = "exp")]
+    pub expires: i64,
+    #[serde(rename = "jti")]
+    pub id: String,
+}
+
+impl<'de> Deserialize<'de> for Claims {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct ClaimsVisitor;
+        impl<'de> Visitor<'de> for ClaimsVisitor {
+            type Value = Claims;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("a JWT claims object")
+            }
+
+            fn visit_unit<E: serde::de::Error>(self) -> Result<Claims, E> {
+                Ok(Claims::default())
+            }
+
+            fn visit_map<M: MapAccess<'de>>(self, mut map: M) -> Result<Claims, M::Error> {
+                let mut claims = Claims::default();
+                while let Some(key) = map.next_key::<String>()? {
+                    // encoding/json matches tagged struct fields without case,
+                    // and applies duplicate fields in order. Null leaves scalar
+                    // fields unchanged, including a previously decoded value.
+                    match key.to_ascii_lowercase().replace('ſ', "s").as_str() {
+                        "sub" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.subject = value;
+                            }
+                        }
+                        "email" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.email = value;
+                            }
+                        }
+                        "iss" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.issuer = value;
+                            }
+                        }
+                        "iat" => {
+                            if let Some(value) = map.next_value::<Option<i64>>()? {
+                                claims.issued = value;
+                            }
+                        }
+                        "exp" => {
+                            if let Some(value) = map.next_value::<Option<i64>>()? {
+                                claims.expires = value;
+                            }
+                        }
+                        "jti" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.id = value;
+                            }
+                        }
+                        _ => {
+                            map.next_value::<serde::de::IgnoredAny>()?;
+                        }
+                    }
+                }
+                Ok(claims)
+            }
+        }
+        deserializer.deserialize_any(ClaimsVisitor)
+    }
+}
+
+#[derive(Clone)]
+pub struct TokenManager {
+    secret: Vec<u8>,
+    access_ttl: TimeDelta,
+}
+
+impl TokenManager {
+    pub fn new(secret: &str, access_ttl: TimeDelta) -> Self {
+        Self {
+            secret: secret.as_bytes().to_vec(),
+            access_ttl,
+        }
+    }
+
+    pub fn access_token(
+        &self,
+        user_id: &str,
+        email: &str,
+    ) -> Result<(String, DateTime<Utc>), String> {
+        let now = Utc::now();
+        let expires = now + self.access_ttl;
+        let claims = Claims {
+            subject: user_id.into(),
+            email: email.into(),
+            issuer: "balance-server".into(),
+            issued: now.timestamp(),
+            expires: expires.timestamp(),
+            id: random_token(16)?,
+        };
+        let payload = serde_json::to_vec(&claims).map_err(|error| error.to_string())?;
+        let unsigned = format!(
+            "{}.{}",
+            URL_SAFE_NO_PAD.encode(br#"{"alg":"HS256","typ":"JWT"}"#),
+            URL_SAFE_NO_PAD.encode(payload)
+        );
+        Ok((format!("{}.{}", unsigned, self.sign(&unsigned)), expires))
+    }
+
+    pub fn verify(&self, token: &str) -> Result<Claims, String> {
+        let parts: Vec<_> = token.split('.').collect();
+        if parts.len() != 3 {
+            return Err("invalid token".into());
+        }
+        let unsigned = format!("{}.{}", parts[0], parts[1]);
+        // Go compares the encoded signature, so reject noncanonical encodings too.
+        let expected = self.sign(&unsigned);
+        let mut mac =
+            Hmac::<Sha256>::new_from_slice(&self.secret).expect("HMAC accepts any key size");
+        mac.update(parts[2].as_bytes());
+        let signature_tag = mac.finalize().into_bytes();
+        let mut expected_mac =
+            Hmac::<Sha256>::new_from_slice(&self.secret).expect("HMAC accepts any key size");
+        expected_mac.update(expected.as_bytes());
+        if expected_mac.verify_slice(&signature_tag).is_err() {
+            return Err("invalid token signature".into());
+        }
+        let payload =
+            decode_go_base64(parts[1], true).map_err(|_| "invalid token payload".to_string())?;
+        let claims: Claims = serde_json::from_slice(&crate::go_json::normalize(&payload))
+            .map_err(|_| "invalid token claims".to_string())?;
+        if claims.issuer != "balance-server"
+            || claims.subject.is_empty()
+            || Utc::now().timestamp() >= claims.expires
+        {
+            return Err("expired or invalid token".into());
+        }
+        Ok(claims)
+    }
+
+    fn sign(&self, value: &str) -> String {
+        let mut mac =
+            Hmac::<Sha256>::new_from_slice(&self.secret).expect("HMAC accepts any key size");
+        mac.update(value.as_bytes());
+        URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn jwt_claims_follow_go_scalar_and_duplicate_field_rules() {
+        let manager = TokenManager::new("secret", TimeDelta::minutes(1));
+        let claims = r#"{"SUB":"old","sub":"user","sub":null,"ISS":"balance-server","EXP":4102444800,"exp":null,"EMAIL":null,"iat":null,"jti":"id","JTI":null}"#;
+        // The Go verifier does not inspect or decode the JWT header.
+        let unsigned = format!("ignored-header.{}", URL_SAFE_NO_PAD.encode(claims));
+        let token = format!("{}.{}", unsigned, manager.sign(&unsigned));
+        let actual = manager.verify(&token).unwrap();
+        assert_eq!(actual.subject, "user");
+        assert_eq!(actual.email, "");
+        assert_eq!(actual.expires, 4_102_444_800);
+        assert_eq!(actual.id, "id");
+        for invalid in [r#"{"sub":1}"#, r#"{"iat":1.0}"#, r#"[]"#] {
+            assert!(serde_json::from_str::<Claims>(invalid).is_err());
+        }
+    }
+
+    #[test]
+    fn jwt_claims_replace_malformed_unicode_as_go_does() {
+        let manager = TokenManager::new("secret", TimeDelta::minutes(1));
+        for payload in [
+            br#"{"sub":"\ud800","iss":"balance-server","exp":4102444800}"#.as_slice(),
+            b"{\"sub\":\"\xff\",\"iss\":\"balance-server\",\"exp\":4102444800}".as_slice(),
+        ] {
+            let unsigned = format!("ignored.{}", URL_SAFE_NO_PAD.encode(payload));
+            let token = format!("{}.{}", unsigned, manager.sign(&unsigned));
+            assert_eq!(manager.verify(&token).unwrap().subject, "�");
+        }
+    }
+
+    #[test]
+    fn go_base64_ignores_line_breaks_and_unused_bits_only() {
+        assert_eq!(decode_go_base64("Zh\r\n", true).unwrap(), b"f");
+        assert_eq!(decode_go_base64("Zh\r\n", false).unwrap(), b"f");
+        for invalid in ["Zg=", "Z g", "Zg\t", "Z"] {
+            assert!(decode_go_base64(invalid, true).is_err());
+        }
+        assert_eq!(
+            password_parameters("m=65536,t=3,p=2suffix"),
+            Some((65536, 3, 2))
+        );
+        assert_eq!(password_parameters("m=65536,t= 3,p=2"), Some((65536, 3, 2)));
+        assert_eq!(password_parameters("m=65536,t=3,p=256"), None);
+        assert_eq!(password_parameters("m=+65536,t=3,p=2"), None);
+    }
+
+    #[test]
+    fn accepts_existing_password_hash_with_go_parameters() {
+        // Fixed salt and an independently computed libargon2 reference result.
+        // Go's argon2.IDKey uses this same Argon2id v19 representation.
+        let encoded = "$argon2id$v=19$m=65536,t=3,p=2$MDEyMzQ1Njc4OWFiY2RlZg$0R79n3aEb762TWx/8plp8QrcDJd8M65F5ccBf1ugp4Y";
+        assert!(verify_password(encoded, "Go-compatible-✓"));
+        assert!(!verify_password(encoded, "Go-compatible-x"));
+        assert_eq!(
+            hash_password("short").unwrap_err(),
+            "password must contain at least 8 characters"
+        );
+        assert_eq!(
+            hash_password(&"x".repeat(1025)).unwrap_err(),
+            "password is too long"
+        );
+        // The original Go implementation measures UTF-8 bytes, not characters.
+        assert!(hash_password("éééé").is_ok());
+        for malformed in [
+            "",
+            "$argon2i$v=19$m=65536,t=3,p=2$c2FsdA$aGFzaA",
+            "$argon2id$v=19$m=999999,t=3,p=2$c2FsdA$aGFzaA",
+            "$argon2id$v=19$m=65536,t=3,p=2$!$!",
+        ] {
+            assert!(!verify_password(malformed, "password"));
+        }
+    }
+
+    #[test]
+    fn accepts_go_jwt_wire_format_and_preserves_refresh_hashes() {
+        // Signed independently using Go's exact header, claims and HS256 encoding.
+        let token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJnby11c2VyIiwiZW1haWwiOiJnb0BleGFtcGxlLmNvbSIsImlzcyI6ImJhbGFuY2Utc2VydmVyIiwiaWF0IjoxNzAwMDAwMDAwLCJleHAiOjQxMDI0NDQ4MDAsImp0aSI6ImdvLXNlc3Npb24ifQ.BJrD7QmRGbRhYqS3EI1stCV5AEkocqOcFAJ3yC4MqeI";
+        let manager = TokenManager::new("go-fixture-secret", TimeDelta::minutes(15));
+        let claims = manager.verify(token).unwrap();
+        assert_eq!(claims.subject, "go-user");
+        assert_eq!(claims.email, "go@example.com");
+        assert_eq!(claims.expires, 4_102_444_800);
+        assert_eq!(claims.id, "go-session");
+        assert!(manager.verify(&format!("{token}=")).is_err());
+        assert!(
+            manager
+                .verify(&token.replace("eyJzdWI", "eyJzdWJ"))
+                .is_err()
+        );
+        assert_eq!(
+            token_hash("existing-go-refresh-token"),
+            "66e9debb6786892dfecaec6bff6761635629efae55baaeb78645820772911605"
+        );
+        assert_eq!(
+            URL_SAFE_NO_PAD
+                .decode(random_token(48).unwrap())
+                .unwrap()
+                .len(),
+            48
+        );
+        let id = identifier().unwrap();
+        assert_eq!(id.len(), 32);
+        assert!(
+            id.bytes()
+                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+        );
+    }
+
+    #[test]
+    fn passwords_and_tokens_round_trip() {
+        let encoded = hash_password("password123").unwrap();
+        assert!(encoded.starts_with("$argon2id$v=19$m=65536,t=3,p=2$"));
+        assert!(verify_password(&encoded, "password123"));
+        assert!(!verify_password(&encoded, "wrong"));
+        let manager = TokenManager::new("test secret", TimeDelta::minutes(15));
+        let (token, _) = manager.access_token("user", "user@example.com").unwrap();
+        assert_eq!(manager.verify(&token).unwrap().subject, "user");
+        assert!(
+            TokenManager::new("other", TimeDelta::minutes(15))
+                .verify(&token)
+                .is_err()
+        );
+        let (expired, _) = TokenManager::new("test secret", TimeDelta::seconds(-1))
+            .access_token("user", "email")
+            .unwrap();
+        assert!(manager.verify(&expired).is_err());
+    }
+}
@@ -0,0 +1,160 @@
+use chrono::TimeDelta;
+use std::env;
+
+#[derive(Clone, Debug)]
+pub struct Config {
+    pub address: String,
+    pub database_path: String,
+    pub jwt_secret: String,
+    pub access_ttl: TimeDelta,
+    pub refresh_ttl: TimeDelta,
+    pub allow_registration: bool,
+    pub cors_origins: Vec<String>,
+}
+
+impl Config {
+    pub fn load() -> Result<Self, String> {
+        let access_ttl = duration("BALANCE_ACCESS_TTL", TimeDelta::minutes(15))?;
+        let refresh_ttl = duration("BALANCE_REFRESH_TTL", TimeDelta::days(30))?;
+        let allow_registration = match value("BALANCE_ALLOW_REGISTRATION").as_str() {
+            "" | "1" | "t" | "T" | "TRUE" | "true" | "True" => true,
+            "0" | "f" | "F" | "FALSE" | "false" | "False" => false,
+            _ => return Err("BALANCE_ALLOW_REGISTRATION must be true or false".into()),
+        };
+        let jwt_secret = value("BALANCE_JWT_SECRET");
+        if jwt_secret.len() < 32 {
+            return Err("BALANCE_JWT_SECRET must contain at least 32 characters".into());
+        }
+        Ok(Self {
+            address: fallback("BALANCE_ADDR", ":8080"),
+            database_path: fallback("BALANCE_DB_PATH", "./data/balance.db"),
+            jwt_secret,
+            access_ttl,
+            refresh_ttl,
+            allow_registration,
+            cors_origins: value("BALANCE_CORS_ORIGINS")
+                .split(',')
+                .map(str::trim)
+                .filter(|part| !part.is_empty())
+                .map(str::to_owned)
+                .collect(),
+        })
+    }
+}
+
+fn value(key: &str) -> String {
+    env::var(key).unwrap_or_default().trim().to_owned()
+}
+fn fallback(key: &str, default: &str) -> String {
+    let value = value(key);
+    if value.is_empty() {
+        default.to_owned()
+    } else {
+        value
+    }
+}
+fn duration(key: &str, default: TimeDelta) -> Result<TimeDelta, String> {
+    let value = value(key);
+    if value.is_empty() {
+        return Ok(default);
+    }
+    parse_go_duration(&value)
+        .ok_or_else(|| format!("{key} must be a Go duration such as 15m or 720h"))
+}
+
+/// Parse Go's signed, nanosecond-precision duration syntax and int64 range.
+pub fn parse_go_duration(input: &str) -> Option<TimeDelta> {
+    let (negative, mut remaining) = match input.as_bytes().first()? {
+        b'-' => (true, &input[1..]),
+        b'+' => (false, &input[1..]),
+        _ => (false, input),
+    };
+    if remaining == "0" {
+        return Some(TimeDelta::zero());
+    }
+    if remaining.is_empty() {
+        return None;
+    }
+    let mut total = 0i128;
+    while !remaining.is_empty() {
+        let integer_len = remaining.bytes().take_while(u8::is_ascii_digit).count();
+        let integer = if integer_len == 0 {
+            0
+        } else {
+            remaining[..integer_len].parse::<i128>().ok()?
+        };
+        remaining = &remaining[integer_len..];
+        let mut fraction = 0i128;
+        let mut scale = 1i128;
+        let mut fraction_len = 0;
+        if remaining.starts_with('.') {
+            remaining = &remaining[1..];
+            fraction_len = remaining.bytes().take_while(u8::is_ascii_digit).count();
+            // Beyond 18 decimal places no supported unit can affect nanoseconds.
+            for digit in remaining.bytes().take(fraction_len.min(18)) {
+                fraction = fraction * 10 + i128::from(digit - b'0');
+                scale *= 10;
+            }
+            remaining = &remaining[fraction_len..];
+        }
+        if integer_len == 0 && fraction_len == 0 {
+            return None;
+        }
+        let (unit_text, unit) = [
+            ("ns", 1i128),
+            ("us", 1_000),
+            ("µs", 1_000),
+            ("μs", 1_000),
+            ("ms", 1_000_000),
+            ("s", 1_000_000_000),
+            ("m", 60_000_000_000),
+            ("h", 3_600_000_000_000),
+        ]
+        .into_iter()
+        .find(|(unit, _)| remaining.starts_with(unit))?;
+        remaining = &remaining[unit_text.len()..];
+        total = total
+            .checked_add(integer.checked_mul(unit)?)?
+            .checked_add(fraction * unit / scale)?;
+        if total > i128::from(i64::MAX) + i128::from(negative) {
+            return None;
+        }
+    }
+    let nanos = i64::try_from(if negative { -total } else { total }).ok()?;
+    Some(TimeDelta::nanoseconds(nanos))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn duration_matches_go_syntax_and_range() {
+        assert_eq!(parse_go_duration("1h30m"), Some(TimeDelta::minutes(90)));
+        assert_eq!(
+            parse_go_duration("-.5s"),
+            Some(TimeDelta::milliseconds(-500))
+        );
+        assert_eq!(parse_go_duration("1µs"), Some(TimeDelta::microseconds(1)));
+        assert_eq!(parse_go_duration("1.s"), Some(TimeDelta::seconds(1)));
+        assert_eq!(parse_go_duration("0"), Some(TimeDelta::zero()));
+        assert_eq!(
+            parse_go_duration("9223372036854775807ns"),
+            Some(TimeDelta::nanoseconds(i64::MAX))
+        );
+        assert_eq!(
+            parse_go_duration("-9223372036854775808ns"),
+            Some(TimeDelta::nanoseconds(i64::MIN))
+        );
+        for invalid in [
+            "",
+            "1d",
+            "1",
+            "9223372036854775808ns",
+            "1h 2m",
+            ".s",
+            "--1s",
+        ] {
+            assert!(parse_go_duration(invalid).is_none(), "{invalid}");
+        }
+    }
+}
@@ -0,0 +1,185 @@
+use rusqlite::Connection;
+use std::{error::Error, path::Path, time::Duration};
+
+/// Open the existing Go server's SQLite database without changing its schema.
+pub fn open(path: &str) -> Result<Connection, Box<dyn Error + Send + Sync>> {
+    if let Some(parent) = Path::new(path)
+        .parent()
+        .filter(|p| !p.as_os_str().is_empty())
+    {
+        let mut builder = std::fs::DirBuilder::new();
+        builder.recursive(true);
+        // Match Go's MkdirAll(..., 0750): the database directory must not be
+        // accessible to other local users under the usual 022 umask.
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::DirBuilderExt;
+            builder.mode(0o750);
+        }
+        builder.create(parent)?;
+    }
+    let db = Connection::open(path)?;
+    db.pragma_update(None, "foreign_keys", true)?;
+    db.pragma_update(None, "journal_mode", "WAL")?;
+    db.busy_timeout(Duration::from_secs(5))?;
+    migrate(&db)?;
+    Ok(db)
+}
+
+pub fn migrate(db: &Connection) -> rusqlite::Result<()> {
+    db.execute_batch(
+        "CREATE TABLE IF NOT EXISTS users (
+  id TEXT PRIMARY KEY,
+  email TEXT NOT NULL UNIQUE COLLATE NOCASE,
+  password_hash TEXT NOT NULL,
+  created_at TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS refresh_tokens (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  token_hash TEXT NOT NULL UNIQUE,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  expires_at TEXT NOT NULL,
+  created_at TEXT NOT NULL,
+  revoked_at TEXT
+);
+CREATE INDEX IF NOT EXISTS refresh_tokens_user_idx ON refresh_tokens(user_id);
+CREATE TABLE IF NOT EXISTS records (
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  entity TEXT NOT NULL,
+  record_id TEXT NOT NULL,
+  payload TEXT,
+  updated_at TEXT NOT NULL,
+  deleted INTEGER NOT NULL DEFAULT 0,
+  version INTEGER NOT NULL,
+  PRIMARY KEY (user_id, entity, record_id)
+);
+CREATE TABLE IF NOT EXISTS changes (
+  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  entity TEXT NOT NULL,
+  record_id TEXT NOT NULL,
+  payload TEXT,
+  updated_at TEXT NOT NULL,
+  deleted INTEGER NOT NULL,
+  version INTEGER NOT NULL
+);
+CREATE INDEX IF NOT EXISTS changes_user_sequence_idx ON changes(user_id, sequence);",
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn schema_is_repeatable_and_enforces_existing_constraints() {
+        let db = open(":memory:").unwrap();
+        migrate(&db).unwrap();
+        db.execute(
+            "INSERT INTO users VALUES ('u','Test@example.com','hash','now')",
+            [],
+        )
+        .unwrap();
+        assert!(
+            db.execute(
+                "INSERT INTO users VALUES ('v','test@example.com','hash','now')",
+                []
+            )
+            .is_err()
+        );
+        assert!(
+            db.execute(
+                "INSERT INTO records VALUES ('missing','category','c',NULL,'now',1,1)",
+                []
+            )
+            .is_err()
+        );
+        db.execute("INSERT INTO refresh_tokens(token_hash,user_id,expires_at,created_at) VALUES ('hash','u','then','now')", []).unwrap();
+        db.execute("DELETE FROM users WHERE id='u'", []).unwrap();
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM refresh_tokens", [], |r| r
+                .get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+    }
+
+    #[test]
+    fn opens_populated_go_schema_without_changing_rows_or_sequence() {
+        let directory = std::env::temp_dir().join(format!(
+            "balance-go-db-{}",
+            crate::auth::identifier().unwrap()
+        ));
+        std::fs::create_dir(&directory).unwrap();
+        let path = directory.join("existing.sqlite");
+        {
+            let db = Connection::open(&path).unwrap();
+            // Use the original schema, so a divergence between implementations
+            // is caught even when Rust's migration is internally consistent.
+            let go_source = include_str!("../internal/database/database.go");
+            let schema = go_source
+                .split("const schema = `")
+                .nth(1)
+                .unwrap()
+                .split('`')
+                .next()
+                .unwrap();
+            db.execute_batch(schema).unwrap();
+            db.execute_batch(
+                "INSERT INTO users VALUES ('go-user','go@example.com','existing hash','2026-01-01T00:00:00.123456789Z');
+                 INSERT INTO refresh_tokens(id,token_hash,user_id,expires_at,created_at) VALUES (7,'existing-token-hash','go-user','2099-01-01T00:00:00Z','2026-01-01T00:00:00Z');
+                 INSERT INTO records VALUES ('go-user','transaction','record', '{\"amount\":12.5}', '2026-01-01T00:00:00.123456789Z',0,41);
+                 INSERT INTO changes VALUES (41,'go-user','transaction','record','{\"amount\":12.5}','2026-01-01T00:00:00.123456789Z',0,41);"
+            ).unwrap();
+        }
+        {
+            let db = open(path.to_str().unwrap()).unwrap();
+            let user: (String, String) = db
+                .query_row(
+                    "SELECT password_hash,created_at FROM users WHERE id='go-user'",
+                    [],
+                    |row| Ok((row.get(0)?, row.get(1)?)),
+                )
+                .unwrap();
+            assert_eq!(
+                user,
+                (
+                    "existing hash".into(),
+                    "2026-01-01T00:00:00.123456789Z".into()
+                )
+            );
+            assert_eq!(
+                db.query_row(
+                    "SELECT id FROM refresh_tokens WHERE revoked_at IS NULL",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )
+                .unwrap(),
+                7
+            );
+            assert_eq!(
+                db.query_row("SELECT payload FROM records", [], |r| r.get::<_, String>(0))
+                    .unwrap(),
+                "{\"amount\":12.5}"
+            );
+            db.execute("INSERT INTO changes(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES ('go-user','transaction','next',NULL,'now',1,42)", []).unwrap();
+            assert_eq!(db.last_insert_rowid(), 42);
+            assert_eq!(
+                db.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0))
+                    .unwrap(),
+                1
+            );
+            assert_eq!(
+                db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
+                    .unwrap(),
+                "wal"
+            );
+            assert_eq!(
+                db.query_row("PRAGMA busy_timeout", [], |r| r.get::<_, i64>(0))
+                    .unwrap(),
+                5000
+            );
+        }
+        std::fs::remove_dir_all(directory).unwrap();
+    }
+}
@@ -0,0 +1,326 @@
+//! Compatibility with encoding/json's replacement of malformed Unicode in
+//! decoded JSON strings. Syntax errors remain errors for the JSON decoder.
+//!
+//! Apply this only after enforcing limits on the original request bytes.
+//! RawMessage/RawValue fields must retain their original input separately:
+//! Go does not replace malformed Unicode inside a RawMessage.
+
+/// Replace malformed UTF-8 and unpaired UTF-16 escapes inside JSON strings.
+/// Valid bytes, whitespace and escape spellings are preserved exactly.
+pub fn normalize(input: &[u8]) -> Vec<u8> {
+    let mut output = Vec::with_capacity(input.len());
+    let mut index = 0;
+    let mut in_string = false;
+    while index < input.len() {
+        let byte = input[index];
+        if !in_string {
+            output.push(byte);
+            in_string = byte == b'"';
+            index += 1;
+            continue;
+        }
+        match byte {
+            b'"' => {
+                output.push(byte);
+                in_string = false;
+                index += 1;
+            }
+            b'\\' => {
+                if let Some(code) = unicode_escape(&input[index..]) {
+                    if (0xD800..=0xDBFF).contains(&code) {
+                        if unicode_escape(&input[index + 6..])
+                            .is_some_and(|low| (0xDC00..=0xDFFF).contains(&low))
+                        {
+                            output.extend_from_slice(&input[index..index + 12]);
+                            index += 12;
+                            continue;
+                        }
+                        output.extend_from_slice(b"\\uFFFD");
+                    } else if (0xDC00..=0xDFFF).contains(&code) {
+                        output.extend_from_slice(b"\\uFFFD");
+                    } else {
+                        output.extend_from_slice(&input[index..index + 6]);
+                    }
+                    index += 6;
+                } else {
+                    // Skip an escaped quote/backslash as one unit; invalid
+                    // escape syntax is left intact for serde to reject.
+                    let end = (index + 2).min(input.len());
+                    output.extend_from_slice(&input[index..end]);
+                    index = end;
+                }
+            }
+            0..=0x7f => {
+                output.push(byte);
+                index += 1;
+            }
+            _ => {
+                let length = match byte {
+                    0xC2..=0xDF => 2,
+                    0xE0..=0xEF => 3,
+                    0xF0..=0xF4 => 4,
+                    _ => 0,
+                };
+                if length != 0
+                    && index + length <= input.len()
+                    && std::str::from_utf8(&input[index..index + length]).is_ok()
+                {
+                    output.extend_from_slice(&input[index..index + length]);
+                    index += length;
+                } else {
+                    // Go's utf8.DecodeRune consumes one byte for every error,
+                    // including each byte of an incomplete sequence.
+                    output.extend_from_slice("\u{FFFD}".as_bytes());
+                    index += 1;
+                }
+            }
+        }
+    }
+    output
+}
+
+fn unicode_escape(input: &[u8]) -> Option<u16> {
+    if input.len() < 6 || &input[..2] != b"\\u" {
+        return None;
+    }
+    input[2..6].iter().try_fold(0u16, |value, byte| {
+        let digit = match byte {
+            b'0'..=b'9' => byte - b'0',
+            b'a'..=b'f' => byte - b'a' + 10,
+            b'A'..=b'F' => byte - b'A' + 10,
+            _ => return None,
+        };
+        Some(value * 16 + u16::from(digit))
+    })
+}
+
+/// Replace named raw fields with numbered JSON strings before scalar decoding.
+/// The returned raw values retain the original byte spelling and whitespace.
+pub fn extract_raw_fields(input: &[u8], field: &str) -> Result<(Vec<u8>, Vec<Vec<u8>>), ()> {
+    let start = skip_space(input, 0);
+    let end = value_end(input, start)?;
+    let mut output = Vec::with_capacity(input.len());
+    let mut raw = Vec::new();
+    let mut copied = 0;
+    let mut index = start;
+    while index < end {
+        if input[index] != b'"' {
+            index += 1;
+            continue;
+        }
+        let string_end = quoted_end(input, index)?;
+        let colon = skip_space(input, string_end);
+        if input.get(colon) == Some(&b':') {
+            let key: String =
+                serde_json::from_slice(&normalize(&input[index..string_end])).map_err(|_| ())?;
+            if key.to_ascii_lowercase().replace('ſ', "s") == field {
+                let value_start = skip_space(input, colon + 1);
+                let value_end = value_end(input, value_start)?;
+                let bytes = &input[value_start..value_end];
+                // Validate raw syntax without converting numbers or strings.
+                serde_json::from_slice::<Box<serde_json::value::RawValue>>(&normalize(bytes))
+                    .map_err(|_| ())?;
+                output.extend_from_slice(&input[copied..value_start]);
+                output.extend_from_slice(format!("\"{}\"", raw.len()).as_bytes());
+                raw.push(bytes.to_vec());
+                copied = value_end;
+                index = value_end;
+                continue;
+            }
+        }
+        index = string_end;
+    }
+    output.extend_from_slice(&input[copied..]);
+    Ok((output, raw))
+}
+
+fn skip_space(input: &[u8], mut index: usize) -> usize {
+    while input
+        .get(index)
+        .is_some_and(|b| matches!(b, b' ' | b'\r' | b'\n' | b'\t'))
+    {
+        index += 1;
+    }
+    index
+}
+
+fn quoted_end(input: &[u8], mut index: usize) -> Result<usize, ()> {
+    index += 1;
+    while index < input.len() {
+        match input[index] {
+            b'"' => return Ok(index + 1),
+            b'\\' => index += 2,
+            _ => index += 1,
+        }
+    }
+    Err(())
+}
+
+fn value_end(input: &[u8], start: usize) -> Result<usize, ()> {
+    match input.get(start).ok_or(())? {
+        b'"' => quoted_end(input, start),
+        b'[' | b'{' => {
+            let mut depth = 1usize;
+            let mut index = start + 1;
+            while index < input.len() {
+                match input[index] {
+                    b'"' => {
+                        index = quoted_end(input, index)?;
+                        continue;
+                    }
+                    b'[' | b'{' => {
+                        depth += 1;
+                        if depth > 10_000 {
+                            return Err(());
+                        }
+                    }
+                    b']' | b'}' => {
+                        depth -= 1;
+                        if depth == 0 {
+                            return Ok(index + 1);
+                        }
+                    }
+                    _ => {}
+                }
+                index += 1;
+            }
+            Err(())
+        }
+        _ => {
+            let length = input[start..]
+                .iter()
+                .position(|b| matches!(b, b' ' | b'\r' | b'\n' | b'\t' | b',' | b']' | b'}'))
+                .unwrap_or(input.len() - start);
+            if length == 0 {
+                Err(())
+            } else {
+                Ok(start + length)
+            }
+        }
+    }
+}
+
+/// Go's encoding/json compacts RawMessage and escapes HTML-sensitive bytes,
+/// preserving duplicate keys, number spellings and malformed Unicode bytes.
+pub fn compact_raw(input: &[u8]) -> Vec<u8> {
+    let mut result = Vec::with_capacity(input.len());
+    let mut in_string = false;
+    let mut escaped = false;
+    let mut index = 0;
+    while index < input.len() {
+        let byte = input[index];
+        if !in_string && matches!(byte, b' ' | b'\n' | b'\r' | b'\t') {
+            index += 1;
+            continue;
+        }
+        if in_string && !escaped {
+            let replacement: Option<&[u8]> = match byte {
+                b'<' => Some(b"\\u003c"),
+                b'>' => Some(b"\\u003e"),
+                b'&' => Some(b"\\u0026"),
+                _ => None,
+            };
+            if let Some(value) = replacement {
+                result.extend_from_slice(value);
+                index += 1;
+                continue;
+            }
+            if input[index..].starts_with(&[0xE2, 0x80, 0xA8])
+                || input[index..].starts_with(&[0xE2, 0x80, 0xA9])
+            {
+                result.extend_from_slice(if input[index + 2] == 0xA8 {
+                    b"\\u2028"
+                } else {
+                    b"\\u2029"
+                });
+                index += 3;
+                continue;
+            }
+        }
+        result.push(byte);
+        if escaped {
+            escaped = false;
+        } else if in_string && byte == b'\\' {
+            escaped = true;
+        } else if byte == b'"' {
+            in_string = !in_string;
+        }
+        index += 1;
+    }
+    result
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn preserves_valid_json_and_literal_backslashes() {
+        let input =
+            br#" {"escaped":"\\ud800", "pair":"\ud83d\ude00", "quote":"\"", "slash":"\\"} "#;
+        assert_eq!(normalize(input), input);
+        serde_json::from_slice::<serde_json::Value>(&normalize(input)).unwrap();
+        let unicode = "{\"text\":\"Привет 🌍\"}".as_bytes();
+        assert_eq!(normalize(unicode), unicode);
+    }
+
+    #[test]
+    fn replaces_unpaired_surrogates_without_consuming_following_escapes() {
+        let input = br#"["\ud800", "\udc00", "\ud800\u0061", "\ud800\ud800\udc00"]"#;
+        let value: Vec<String> = serde_json::from_slice(&normalize(input)).unwrap();
+        assert_eq!(value, ["�", "�", "�a", "�𐀀"]);
+    }
+
+    #[test]
+    fn replaces_each_malformed_utf8_byte_as_go_does() {
+        for (input, expected) in [
+            (b"\"\xff\"".as_slice(), "�"),
+            (b"\"\xe1\x80\"".as_slice(), "��"),
+            (b"\"\xed\xa0\x80\"".as_slice(), "���"),
+            (b"\"\xf4\x90\x80\x80\"".as_slice(), "����"),
+        ] {
+            let decoded: String = serde_json::from_slice(&normalize(input)).unwrap();
+            assert_eq!(decoded, expected);
+        }
+    }
+
+    #[test]
+    fn leaves_invalid_json_syntax_rejected() {
+        for input in [
+            br#""\uZZZZ""#.as_slice(),
+            br#""\x""#.as_slice(),
+            b"\"raw\nnewline\"".as_slice(),
+            b"{\xff:1}".as_slice(),
+        ] {
+            assert!(serde_json::from_slice::<serde_json::Value>(&normalize(input)).is_err());
+        }
+    }
+
+    #[test]
+    fn raw_extraction_preserves_bytes_and_ignores_trailing_json() {
+        let input = b"{\"payload\": { \"x\":\"\\ud800\", \"x\":\"\xff\" },\"PAYLOAD\":null} {\"payload\":not-json}";
+        let (masked, values) = extract_raw_fields(input, "payload").unwrap();
+        assert_eq!(
+            values,
+            [
+                b"{ \"x\":\"\\ud800\", \"x\":\"\xff\" }".to_vec(),
+                b"null".to_vec()
+            ]
+        );
+        assert_eq!(
+            masked,
+            b"{\"payload\": \"0\",\"PAYLOAD\":\"1\"} {\"payload\":not-json}"
+        );
+        assert!(extract_raw_fields(br#"{"payload":{"bad":}}"#, "payload").is_err());
+        assert!(extract_raw_fields(br#"{"payload":[1,]}"#, "payload").is_err());
+    }
+
+    #[test]
+    fn raw_compaction_matches_go_html_escaping_without_decoding_strings() {
+        let input = "{ \"raw\":\"<&>\u{2028}\u{2029}\", \"escaped\":\"\\u003c\", \"dup\":1, \"dup\":2, \"n\":1.00e+999 }";
+        assert_eq!(
+            compact_raw(input.as_bytes()),
+            br#"{"raw":"\u003c\u0026\u003e\u2028\u2029","escaped":"\u003c","dup":1,"dup":2,"n":1.00e+999}"#
+        );
+    }
+}
@@ -0,0 +1,49 @@
+mod auth;
+mod config;
+mod database;
+mod go_json;
+mod runtime;
+mod server;
+mod sync;
+
+use std::sync::{Arc, Mutex};
+fn main() {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .expect("create async runtime");
+    let result = runtime.block_on(run());
+    // Connection draining already has a ten-second budget. Blocking database
+    // or password jobs must not extend process shutdown beyond that budget.
+    runtime.shutdown_timeout(std::time::Duration::ZERO);
+    if let Err(e) = result {
+        eprintln!("Balance server startup failed: {e}");
+        std::process::exit(1);
+    }
+}
+async fn run() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let cfg = config::Config::load()?;
+    let db = database::open(&cfg.database_path)?;
+    let address = cfg.address.clone();
+    let listener = runtime::bind(&address).await?;
+    let state = Arc::new(server::State::new(cfg, Mutex::new(db)));
+    let app = axum::Router::new()
+        .fallback(server::handle)
+        .with_state(state);
+    eprintln!("Balance server listening on {address}");
+    runtime::serve(listener, app, runtime::Timeouts::default(), async {
+        #[cfg(unix)]
+        {
+            let mut term =
+                tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+                    .expect("signal handler");
+            tokio::select! { _ = tokio::signal::ctrl_c() => {}, _ = term.recv() => {} }
+        }
+        #[cfg(not(unix))]
+        {
+            let _ = tokio::signal::ctrl_c().await;
+        }
+    })
+    .await?;
+    Ok(())
+}
@@ -0,0 +1,550 @@
+//! HTTP/1 transport deadlines matching net/http.Server's socket deadlines.
+use axum::{Router, body::Body, extract::ConnectInfo};
+use http_body::{Body as HttpBody, Frame, SizeHint};
+use hyper::{server::conn::http1, service::service_fn};
+use hyper_util::rt::TokioIo;
+use std::{
+    future::Future,
+    io,
+    net::SocketAddr,
+    pin::Pin,
+    sync::{Arc, Mutex},
+    task::{Context, Poll},
+    time::Duration,
+};
+use tokio::{
+    io::{AsyncRead, AsyncWrite, ReadBuf},
+    net::{TcpListener, TcpStream},
+    sync::watch,
+    task::JoinSet,
+    time::{Instant, Sleep},
+};
+use tower::ServiceExt;
+
+#[derive(Clone, Copy)]
+pub struct Timeouts {
+    pub header: Duration,
+    pub read: Duration,
+    pub write: Duration,
+    pub idle: Duration,
+    pub shutdown: Duration,
+}
+impl Default for Timeouts {
+    fn default() -> Self {
+        Self {
+            header: Duration::from_secs(5),
+            read: Duration::from_secs(15),
+            write: Duration::from_secs(30),
+            idle: Duration::from_secs(60),
+            shutdown: Duration::from_secs(10),
+        }
+    }
+}
+
+struct Deadlines {
+    started: Instant,
+    read: Option<Instant>,
+    write: Option<Instant>,
+    idle: bool,
+    response_done: bool,
+}
+struct DeadlineIo {
+    stream: TcpStream,
+    shared: Arc<Mutex<Deadlines>>,
+    timeouts: Timeouts,
+    read_timer: Pin<Box<Sleep>>,
+    write_timer: Pin<Box<Sleep>>,
+}
+fn expired(timer: &mut Pin<Box<Sleep>>, deadline: Option<Instant>, cx: &mut Context<'_>) -> bool {
+    if let Some(deadline) = deadline {
+        timer.as_mut().reset(deadline);
+        timer.as_mut().poll(cx).is_ready()
+    } else {
+        false
+    }
+}
+fn timed_out() -> io::Error {
+    io::Error::new(io::ErrorKind::TimedOut, "HTTP connection deadline exceeded")
+}
+impl AsyncRead for DeadlineIo {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let deadline = self.shared.lock().unwrap().read;
+        if expired(&mut self.read_timer, deadline, cx) {
+            return Poll::Ready(Err(timed_out()));
+        }
+        let before = buf.filled().len();
+        let result = Pin::new(&mut self.stream).poll_read(cx, buf);
+        if matches!(result, Poll::Ready(Ok(()))) && buf.filled().len() > before {
+            let mut state = self.shared.lock().unwrap();
+            if state.idle {
+                state.idle = false;
+                state.started = Instant::now();
+                state.read = Some(state.started + self.timeouts.header);
+                cx.waker().wake_by_ref();
+            }
+        }
+        result
+    }
+}
+impl AsyncWrite for DeadlineIo {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let deadline = self.shared.lock().unwrap().write;
+        if expired(&mut self.write_timer, deadline, cx) {
+            return Poll::Ready(Err(timed_out()));
+        }
+        Pin::new(&mut self.stream).poll_write(cx, buf)
+    }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let deadline = self.shared.lock().unwrap().write;
+        if expired(&mut self.write_timer, deadline, cx) {
+            return Poll::Ready(Err(timed_out()));
+        }
+        let result = Pin::new(&mut self.stream).poll_flush(cx);
+        if matches!(result, Poll::Ready(Ok(()))) {
+            let mut state = self.shared.lock().unwrap();
+            if state.response_done {
+                state.write = None;
+                state.read = Some(Instant::now() + self.timeouts.idle);
+                state.idle = true;
+                state.response_done = false;
+                cx.waker().wake_by_ref();
+            }
+        }
+        result
+    }
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.stream).poll_shutdown(cx)
+    }
+}
+struct ResponseBody {
+    inner: Body,
+    shared: Arc<Mutex<Deadlines>>,
+}
+impl HttpBody for ResponseBody {
+    type Data = axum::body::Bytes;
+    type Error = axum::Error;
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        let frame = Pin::new(&mut self.inner).poll_frame(cx);
+        if frame.is_ready() && self.inner.is_end_stream() {
+            self.shared.lock().unwrap().response_done = true;
+        }
+        frame
+    }
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+    fn size_hint(&self) -> SizeHint {
+        self.inner.size_hint()
+    }
+}
+
+async fn connection(
+    stream: TcpStream,
+    peer: SocketAddr,
+    app: Router,
+    timeouts: Timeouts,
+    mut shutdown: watch::Receiver<bool>,
+) {
+    let now = Instant::now();
+    let state = Arc::new(Mutex::new(Deadlines {
+        started: now,
+        read: Some(now + timeouts.header),
+        write: None,
+        idle: false,
+        response_done: false,
+    }));
+    let io = DeadlineIo {
+        stream,
+        shared: Arc::clone(&state),
+        timeouts,
+        read_timer: Box::pin(tokio::time::sleep(timeouts.header)),
+        write_timer: Box::pin(tokio::time::sleep(timeouts.write)),
+    };
+    let service = service_fn(move |mut request: hyper::Request<hyper::body::Incoming>| {
+        request.extensions_mut().insert(ConnectInfo(peer));
+        let app = app.clone();
+        let state = Arc::clone(&state);
+        {
+            let mut deadlines = state.lock().unwrap();
+            if deadlines.idle {
+                deadlines.started = Instant::now();
+            }
+            deadlines.idle = false;
+            deadlines.response_done = false;
+            deadlines.read = Some(deadlines.started + timeouts.read);
+            deadlines.write = Some(Instant::now() + timeouts.write);
+        }
+        async move {
+            let response = app.oneshot(request.map(Body::new)).await?;
+            let (parts, body) = response.into_parts();
+            if body.is_end_stream() {
+                state.lock().unwrap().response_done = true;
+            }
+            Ok::<_, std::convert::Infallible>(hyper::Response::from_parts(
+                parts,
+                ResponseBody {
+                    inner: body,
+                    shared: state,
+                },
+            ))
+        }
+    });
+    let conn = http1::Builder::new().serve_connection(TokioIo::new(io), service);
+    tokio::pin!(conn);
+    tokio::select! {
+        _ = &mut conn => {},
+        _ = shutdown.changed() => { conn.as_mut().graceful_shutdown(); let _ = conn.await; }
+    }
+}
+
+/// Go's empty host listens on both IP families where IPv6 is available.
+pub async fn bind(address: &str) -> io::Result<TcpListener> {
+    if address.starts_with(':') {
+        match TcpListener::bind(format!("[::]{address}")).await {
+            Ok(listener) => Ok(listener),
+            Err(_) => TcpListener::bind(format!("0.0.0.0{address}")).await,
+        }
+    } else {
+        TcpListener::bind(address).await
+    }
+}
+
+pub async fn serve(
+    listener: TcpListener,
+    app: Router,
+    timeouts: Timeouts,
+    shutdown: impl Future<Output = ()>,
+) -> io::Result<()> {
+    let (stop, _) = watch::channel(false);
+    let mut connections = JoinSet::new();
+    tokio::pin!(shutdown);
+    loop {
+        tokio::select! {
+            biased;
+            _ = &mut shutdown => break,
+            finished = connections.join_next(), if !connections.is_empty() => { let _ = finished; },
+            accepted = listener.accept() => {
+                let (stream, peer) = accepted?;
+                stream.set_nodelay(true)?;
+                connections.spawn(connection(stream, peer, app.clone(), timeouts, stop.subscribe()));
+            }
+        }
+    }
+    drop(listener);
+    let _ = stop.send(true);
+    if tokio::time::timeout(timeouts.shutdown, async {
+        while connections.join_next().await.is_some() {}
+    })
+    .await
+    .is_err()
+    {
+        connections.abort_all();
+        while connections.join_next().await.is_some() {}
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::routing::{get, post};
+    use tokio::{
+        io::{AsyncReadExt, AsyncWriteExt},
+        sync::oneshot,
+    };
+
+    fn limits() -> Timeouts {
+        Timeouts {
+            header: Duration::from_millis(90),
+            read: Duration::from_millis(220),
+            write: Duration::from_millis(300),
+            idle: Duration::from_millis(500),
+            shutdown: Duration::from_millis(120),
+        }
+    }
+    async fn start(
+        app: Router,
+    ) -> (
+        SocketAddr,
+        oneshot::Sender<()>,
+        tokio::task::JoinHandle<io::Result<()>>,
+    ) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let (stop, stopped) = oneshot::channel();
+        let task = tokio::spawn(serve(listener, app, limits(), async {
+            let _ = stopped.await;
+        }));
+        (addr, stop, task)
+    }
+    async fn closed(stream: &mut TcpStream) {
+        let mut data = [0; 4096];
+        tokio::time::timeout(Duration::from_secs(2), async {
+            match stream.read(&mut data).await {
+                Ok(0) | Err(_) => {}
+                Ok(count) => panic!(
+                    "unexpected response before connection closed: {:?}",
+                    &data[..count]
+                ),
+            }
+        })
+        .await
+        .expect("connection deadline was not enforced");
+    }
+    async fn response(stream: &mut TcpStream) -> String {
+        let mut bytes = Vec::new();
+        while !bytes.ends_with(b"\r\n\r\n") {
+            bytes.push(
+                tokio::time::timeout(Duration::from_secs(2), stream.read_u8())
+                    .await
+                    .unwrap()
+                    .unwrap(),
+            );
+        }
+        let header = String::from_utf8(bytes).unwrap();
+        let len: usize = header
+            .lines()
+            .find_map(|l| {
+                l.to_ascii_lowercase()
+                    .strip_prefix("content-length:")
+                    .map(|s| s.trim().parse().unwrap())
+            })
+            .unwrap_or(0);
+        let mut body = vec![0; len];
+        stream.read_exact(&mut body).await.unwrap();
+        assert_eq!(body, b"ok");
+        header
+    }
+
+    #[tokio::test]
+    async fn incomplete_headers_expire_from_connection_start() {
+        let (addr, stop, task) = start(Router::new().route("/", get(|| async { "ok" }))).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream.write_all(b"GET / HTTP/1.1\r\nHost: ").await.unwrap();
+        let started = Instant::now();
+        closed(&mut stream).await;
+        assert!(started.elapsed() < Duration::from_millis(500));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn body_deadline_includes_header_time_and_is_not_reset_by_progress() {
+        let app = Router::new().route(
+            "/",
+            post(|body: axum::body::Bytes| async move {
+                assert!(body.len() < 10);
+                "ok"
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        let started = Instant::now();
+        stream
+            .write_all(b"POST / HTTP/1.1\r\nHost: localhost\r\n")
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        stream
+            .write_all(b"Content-Length: 10\r\n\r\na")
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        stream.write_all(b"b").await.unwrap();
+        // Axum reports a body-read error, which may generate an HTTP error
+        // response; drain it and wait for the read deadline to close the socket.
+        let mut bytes = Vec::new();
+        tokio::time::timeout(Duration::from_secs(1), stream.read_to_end(&mut bytes))
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(started.elapsed() >= Duration::from_millis(200));
+        assert!(started.elapsed() < Duration::from_millis(400));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn idle_deadline_is_longer_than_header_and_next_headers_get_own_deadline() {
+        let (addr, stop, task) = start(Router::new().route("/", get(|| async { "ok" }))).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        assert!(response(&mut stream).await.starts_with("HTTP/1.1 200"));
+        tokio::time::sleep(Duration::from_millis(180)).await;
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        assert!(response(&mut stream).await.starts_with("HTTP/1.1 200"));
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        stream.write_all(b"GET / HTTP/1.1\r\nHost: ").await.unwrap();
+        let started = Instant::now();
+        closed(&mut stream).await;
+        assert!(started.elapsed() < Duration::from_millis(250));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn idle_keepalive_eventually_closes_without_another_request() {
+        let (addr, stop, task) = start(Router::new().route("/", get(|| async { "ok" }))).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        response(&mut stream).await;
+        let started = Instant::now();
+        closed(&mut stream).await;
+        assert!(started.elapsed() >= Duration::from_millis(450));
+        assert!(started.elapsed() < Duration::from_millis(800));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn responses_cannot_be_written_after_write_deadline() {
+        let app = Router::new().route(
+            "/",
+            get(|| async {
+                tokio::time::sleep(Duration::from_millis(350)).await;
+                "ok"
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        closed(&mut stream).await;
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn shutdown_drains_active_requests_and_bounds_stuck_handlers() {
+        let (entered, notified) = oneshot::channel();
+        let entered = Arc::new(Mutex::new(Some(entered)));
+        let app = Router::new().route(
+            "/",
+            get(move || {
+                let entered = Arc::clone(&entered);
+                async move {
+                    if let Some(entered) = entered.lock().unwrap().take() {
+                        let _ = entered.send(());
+                    }
+                    std::future::pending::<()>().await;
+                    "ok"
+                }
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        notified.await.unwrap();
+        let started = Instant::now();
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+        assert!(started.elapsed() >= Duration::from_millis(100));
+        assert!(started.elapsed() < Duration::from_millis(400));
+        assert!(TcpStream::connect(addr).await.is_err());
+        closed(&mut stream).await;
+    }
+
+    #[tokio::test]
+    async fn shutdown_allows_in_flight_response_to_complete() {
+        let (entered, notified) = oneshot::channel();
+        let entered = Arc::new(Mutex::new(Some(entered)));
+        let app = Router::new().route(
+            "/",
+            get(move || {
+                let entered = Arc::clone(&entered);
+                async move {
+                    if let Some(entered) = entered.lock().unwrap().take() {
+                        let _ = entered.send(());
+                    }
+                    tokio::time::sleep(Duration::from_millis(30)).await;
+                    "ok"
+                }
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        notified.await.unwrap();
+        stop.send(()).unwrap();
+        assert!(response(&mut stream).await.starts_with("HTTP/1.1 200"));
+        task.await.unwrap().unwrap();
+        closed(&mut stream).await;
+    }
+    #[tokio::test]
+    async fn address_binding_supports_hostnames_ip_literals_and_empty_host() {
+        for address in ["127.0.0.1:0", "localhost:0", ":0"] {
+            let listener = bind(address).await.unwrap();
+            let local = listener.local_addr().unwrap();
+            assert_ne!(local.port(), 0);
+            if address == ":0" {
+                let client = TcpStream::connect(("127.0.0.1", local.port()))
+                    .await
+                    .unwrap();
+                let _ = listener.accept().await.unwrap();
+                drop(client);
+            }
+        }
+        // The IPv6 syntax remains accepted on hosts with an IPv6 loopback.
+        match bind("[::1]:0").await {
+            Ok(listener) => assert!(listener.local_addr().unwrap().is_ipv6()),
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
+                ) => {}
+            Err(error) => panic!("IPv6 listener: {error}"),
+        }
+    }
+    #[tokio::test]
+    async fn blocked_socket_writes_expire_without_waiting_for_client_reads() {
+        const LENGTH: usize = 16 * 1024 * 1024;
+        let app = Router::new().route("/", get(|| async { "x".repeat(LENGTH) }));
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        // A response larger than the socket buffer must eventually block.
+        // The write deadline must wake that pending write and close the
+        // connection even though the client has not consumed any response.
+        tokio::time::sleep(Duration::from_millis(400)).await;
+        let mut received = Vec::new();
+        tokio::time::timeout(Duration::from_secs(2), stream.read_to_end(&mut received))
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(received.starts_with(b"HTTP/1.1 200"));
+        assert!(received.len() < LENGTH);
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+}
@@ -0,0 +1,779 @@
+use chrono::{DateTime, Duration, NaiveDate, SecondsFormat, Utc};
+use rusqlite::{Connection, OptionalExtension, Transaction, params};
+use serde::{
+    Deserialize, Deserializer,
+    de::{self, DeserializeSeed, MapAccess, SeqAccess, Visitor},
+};
+#[cfg(test)]
+use serde_json::Value;
+use serde_json::{json, value::RawValue};
+use std::fmt;
+
+pub type SyncError = (u16, &'static str, String);
+
+#[derive(Default)]
+struct Request {
+    cursor: i64,
+    changes: Vec<Change>,
+    pull: Option<bool>,
+}
+
+#[derive(Default)]
+struct Change {
+    entity: String,
+    id: String,
+    deleted: bool,
+    updated_at: Option<DateTime<Utc>>,
+    payload: Option<Vec<u8>>,
+}
+
+// Go's decoder accepts null for ordinary scalar fields, matches keys ignoring
+// case, and lets the last duplicate field win. Keep those wire semantics.
+impl<'de> Deserialize<'de> for Request {
+    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
+        struct RequestVisitor;
+        impl<'de> Visitor<'de> for RequestVisitor {
+            type Value = Request;
+            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                f.write_str("a sync request")
+            }
+            fn visit_unit<E: de::Error>(self) -> Result<Request, E> {
+                Ok(Request::default())
+            }
+            fn visit_map<M: MapAccess<'de>>(self, mut m: M) -> Result<Request, M::Error> {
+                let mut r = Request::default();
+                while let Some(key) = m.next_key::<String>()? {
+                    match fold_key(&key).as_str() {
+                        "cursor" => {
+                            if let Some(v) = m.next_value::<Option<i64>>()? {
+                                r.cursor = v;
+                            }
+                        }
+                        "deviceid" => {
+                            let _: Option<String> = m.next_value()?;
+                        }
+                        "changes" => m.next_value_seed(ChangesSeed(&mut r.changes))?,
+                        "pull" => r.pull = m.next_value()?,
+                        _ => return Err(de::Error::custom("unknown field")),
+                    }
+                }
+                Ok(r)
+            }
+        }
+        d.deserialize_any(RequestVisitor)
+    }
+}
+
+// encoding/json reuses existing slice elements when a duplicate changes field
+// appears. Missing fields in those elements keep their preceding values.
+struct ChangesSeed<'a>(&'a mut Vec<Change>);
+impl<'de> DeserializeSeed<'de> for ChangesSeed<'_> {
+    type Value = ();
+    fn deserialize<D: Deserializer<'de>>(self, d: D) -> Result<(), D::Error> {
+        struct ChangesVisitor<'a>(&'a mut Vec<Change>);
+        impl<'de> Visitor<'de> for ChangesVisitor<'_> {
+            type Value = ();
+            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                f.write_str("an array of changes")
+            }
+            fn visit_unit<E: de::Error>(self) -> Result<(), E> {
+                self.0.clear();
+                Ok(())
+            }
+            fn visit_seq<S: SeqAccess<'de>>(self, mut seq: S) -> Result<(), S::Error> {
+                let mut index = 0;
+                loop {
+                    if index == self.0.len() {
+                        self.0.push(Change::default());
+                    }
+                    if seq
+                        .next_element_seed(ChangeSeed(&mut self.0[index]))?
+                        .is_none()
+                    {
+                        break;
+                    }
+                    index += 1;
+                }
+                self.0.truncate(index);
+                Ok(())
+            }
+        }
+        d.deserialize_any(ChangesVisitor(self.0))
+    }
+}
+
+struct ChangeSeed<'a>(&'a mut Change);
+impl<'de> DeserializeSeed<'de> for ChangeSeed<'_> {
+    type Value = ();
+    fn deserialize<D: Deserializer<'de>>(self, d: D) -> Result<(), D::Error> {
+        struct ChangeVisitor<'a>(&'a mut Change);
+        impl<'de> Visitor<'de> for ChangeVisitor<'_> {
+            type Value = ();
+            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                f.write_str("a sync change")
+            }
+            fn visit_unit<E: de::Error>(self) -> Result<(), E> {
+                Ok(())
+            }
+            fn visit_map<M: MapAccess<'de>>(self, mut m: M) -> Result<(), M::Error> {
+                let c = self.0;
+                while let Some(key) = m.next_key::<String>()? {
+                    match fold_key(&key).as_str() {
+                        "entity" => {
+                            if let Some(v) = m.next_value::<Option<String>>()? {
+                                c.entity = v;
+                            }
+                        }
+                        "id" => {
+                            if let Some(v) = m.next_value::<Option<String>>()? {
+                                c.id = v;
+                            }
+                        }
+                        "deleted" => {
+                            if let Some(v) = m.next_value::<Option<bool>>()? {
+                                c.deleted = v;
+                            }
+                        }
+                        "updatedat" => {
+                            // time.Time.UnmarshalJSON reads the quoted bytes directly;
+                            // JSON escapes inside a timestamp are not unescaped by Go.
+                            let raw = m.next_value::<Box<RawValue>>()?;
+                            if raw.get() != "null" {
+                                let value = raw
+                                    .get()
+                                    .strip_prefix('"')
+                                    .and_then(|v| v.strip_suffix('"'))
+                                    .ok_or_else(|| de::Error::custom("invalid time"))?;
+                                c.updated_at = Some(parse_time(value).map_err(de::Error::custom)?);
+                            }
+                        }
+                        "payload" => {
+                            c.payload =
+                                Some(m.next_value::<Box<RawValue>>()?.get().as_bytes().to_vec());
+                        }
+                        "version" | "sequence" => {
+                            let _: Option<i64> = m.next_value()?;
+                        }
+                        _ => return Err(de::Error::custom("unknown field")),
+                    }
+                }
+                Ok(())
+            }
+        }
+        d.deserialize_any(ChangeVisitor(self.0))
+    }
+}
+
+fn fold_key(s: &str) -> String {
+    // Unicode simple folding in encoding/json also equates long s with S.
+    s.chars()
+        .flat_map(|c| if c == 'ſ' { 's' } else { c }.to_lowercase())
+        .collect()
+}
+
+fn parse_time(s: &str) -> Result<DateTime<Utc>, &'static str> {
+    // Match Go's RFC3339 parser, including its currently enabled fallback:
+    // one-digit hours, comma fractions, offsets up to 24:60, nanosecond
+    // truncation, and rejection of leap seconds and lowercase t/z.
+    let invalid = "invalid RFC3339 timestamp";
+    if !s.is_ascii() {
+        return Err(invalid);
+    }
+    let (date, time) = s.split_once('T').ok_or(invalid)?;
+    let number = |v: &str, width: usize| -> Result<u32, &'static str> {
+        if v.len() != width || !v.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(invalid);
+        }
+        v.parse().map_err(|_| invalid)
+    };
+    if date.len() != 10 {
+        return Err(invalid);
+    }
+    let mut date_parts = date.split('-');
+    let year = number(date_parts.next().ok_or(invalid)?, 4)?;
+    let month = number(date_parts.next().ok_or(invalid)?, 2)?;
+    let day = number(date_parts.next().ok_or(invalid)?, 2)?;
+    if date_parts.next().is_some() {
+        return Err(invalid);
+    }
+    let (clock, offset) = if let Some(clock) = time.strip_suffix('Z') {
+        (clock, 0i64)
+    } else {
+        if time.len() < 6 {
+            return Err(invalid);
+        }
+        let (clock, zone) = time.split_at(time.len() - 6);
+        if !matches!(zone.as_bytes()[0], b'+' | b'-') || zone.as_bytes()[3] != b':' {
+            return Err(invalid);
+        }
+        let hours = number(&zone[1..3], 2)?;
+        let minutes = number(&zone[4..6], 2)?;
+        if hours > 24 || minutes > 60 {
+            return Err(invalid);
+        }
+        let sign = if zone.starts_with('-') { -1 } else { 1 };
+        (clock, sign * i64::from(hours * 3600 + minutes * 60))
+    };
+    let (clock, nanos) = if let Some(index) = clock.find(['.', ',']) {
+        let fraction = &clock[index + 1..];
+        if fraction.is_empty() || !fraction.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(invalid);
+        }
+        let precision = fraction.len().min(9);
+        let nanos = number(&fraction[..precision], precision)? * 10u32.pow((9 - precision) as u32);
+        (&clock[..index], nanos)
+    } else {
+        (clock, 0)
+    };
+    let mut parts = clock.split(':');
+    let hour = parts.next().ok_or(invalid)?;
+    if !(1..=2).contains(&hour.len()) {
+        return Err(invalid);
+    }
+    let hour = number(hour, hour.len())?;
+    let minute = number(parts.next().ok_or(invalid)?, 2)?;
+    let second = number(parts.next().ok_or(invalid)?, 2)?;
+    if parts.next().is_some() || second > 59 {
+        return Err(invalid);
+    }
+    let local = NaiveDate::from_ymd_opt(year as i32, month, day)
+        .and_then(|d| d.and_hms_nano_opt(hour, minute, second, nanos))
+        .ok_or(invalid)?;
+    local
+        .and_utc()
+        .checked_sub_signed(Duration::seconds(offset))
+        .ok_or(invalid)
+}
+
+fn timestamp(t: DateTime<Utc>) -> String {
+    // SecondsFormat::AutoSi pads to 3/6/9 digits, whereas Go strips every zero.
+    let s = t.to_rfc3339_opts(SecondsFormat::Nanos, true);
+    let s = s
+        .trim_end_matches('Z')
+        .trim_end_matches('0')
+        .trim_end_matches('.');
+    format!("{s}Z")
+}
+
+fn failure(status: u16, code: &'static str, message: &str) -> SyncError {
+    (status, code, message.to_owned())
+}
+
+pub fn synchronize_wire(
+    db: &mut Connection,
+    user_id: &str,
+    body: &[u8],
+) -> Result<Vec<u8>, SyncError> {
+    let invalid_json = || failure(400, "invalid_json", "Request body is not valid JSON");
+    let body = &body[..body.len().min(2 << 20)];
+    // The original decoder reads the first JSON value, without requiring EOF.
+    let (masked, raw_payloads) =
+        crate::go_json::extract_raw_fields(body, "payload").map_err(|_| invalid_json())?;
+    let normalized = crate::go_json::normalize(&masked);
+    let mut request = Request::deserialize(&mut serde_json::Deserializer::from_slice(&normalized))
+        .map_err(|_| invalid_json())?;
+    for change in &mut request.changes {
+        if let Some(placeholder) = &change.payload {
+            let index: String = serde_json::from_slice(placeholder).map_err(|_| invalid_json())?;
+            let index: usize = index.parse().map_err(|_| invalid_json())?;
+            change.payload = Some(raw_payloads.get(index).ok_or_else(invalid_json)?.clone());
+        }
+    }
+    if request.cursor < 0 || request.changes.len() > 5000 {
+        return Err(failure(
+            400,
+            "invalid_sync_request",
+            "Cursor or change count is invalid",
+        ));
+    }
+    for c in &request.changes {
+        let message = if !matches!(c.entity.as_str(), "transaction" | "category" | "budget") {
+            Some("entity must be transaction, category, or budget")
+        } else if c.id.is_empty() || c.id.len() > 128 {
+            Some("record id is invalid")
+        } else if c.updated_at.is_none() || c.updated_at == parse_time("0001-01-01T00:00:00Z").ok()
+        {
+            Some("updatedAt is required")
+        } else if c
+            .updated_at
+            .is_some_and(|time| time > Utc::now() + Duration::hours(24))
+        {
+            Some("updatedAt is too far in the future")
+        } else if !c.deleted && c.payload.is_none() {
+            Some("payload is required for active records")
+        } else if c.payload.as_ref().is_some_and(|p| p.len() > 256 * 1024) {
+            Some("payload is too large")
+        } else {
+            None
+        };
+        if let Some(message) = message {
+            return Err(failure(400, "invalid_change", message));
+        }
+    }
+    let tx = db
+        .transaction()
+        .map_err(|_| failure(500, "sync_failed", "Could not begin synchronization"))?;
+    for c in request.changes {
+        apply(&tx, user_id, c)
+            .map_err(|_| failure(500, "sync_failed", "Could not apply changes"))?;
+    }
+    let (changes, cursor, more) = if request.pull.unwrap_or(true) {
+        load(&tx, user_id, request.cursor)
+            .map_err(|_| failure(500, "sync_failed", "Could not load changes"))?
+    } else {
+        (Vec::new(), request.cursor, false)
+    };
+    tx.commit()
+        .map_err(|_| failure(500, "sync_failed", "Could not complete synchronization"))?;
+    let mut response = format!("{{\"cursor\":{cursor},\"changes\":[").into_bytes();
+    for (index, change) in changes.iter().enumerate() {
+        if index != 0 {
+            response.push(b',');
+        }
+        response.extend_from_slice(change);
+    }
+    response.extend_from_slice(
+        format!(
+            "],\"hasMore\":{more},\"serverTime\":{}}}\n",
+            serde_json::to_string(&timestamp(Utc::now())).expect("timestamp JSON")
+        )
+        .as_bytes(),
+    );
+    Ok(response)
+}
+
+#[cfg(test)]
+fn synchronize(db: &mut Connection, user_id: &str, body: &[u8]) -> Result<Value, SyncError> {
+    let response = synchronize_wire(db, user_id, body)?;
+    serde_json::from_slice(&crate::go_json::normalize(&response))
+        .map_err(|_| failure(500, "sync_failed", "Could not serialize synchronization"))
+}
+
+type InternalError = Box<dyn std::error::Error + Send + Sync>;
+
+fn apply(tx: &Transaction<'_>, user: &str, c: Change) -> Result<(), InternalError> {
+    let current: Option<(String, i64)> = tx
+        .query_row(
+            "SELECT updated_at,version FROM records WHERE user_id=? AND entity=? AND record_id=?",
+            params![user, c.entity, c.id],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .optional()?;
+    let updated = c.updated_at.expect("validated timestamp");
+    let mut version = 1;
+    if let Some((time, old_version)) = current {
+        if updated <= parse_time(&time)? {
+            return Ok(());
+        }
+        version = old_version.wrapping_add(1);
+    }
+    let payload = if c.deleted {
+        None
+    } else {
+        c.payload.as_deref()
+    };
+    let time = timestamp(updated);
+    tx.execute("INSERT INTO records(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?) ON CONFLICT(user_id,entity,record_id) DO UPDATE SET payload=excluded.payload,updated_at=excluded.updated_at,deleted=excluded.deleted,version=excluded.version", params![user,c.entity,c.id,payload,time,c.deleted,version])?;
+    tx.execute("INSERT INTO changes(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?)", params![user,c.entity,c.id,payload,time,c.deleted,version])?;
+    Ok(())
+}
+
+fn load(
+    tx: &Transaction<'_>,
+    user: &str,
+    after: i64,
+) -> Result<(Vec<Vec<u8>>, i64, bool), InternalError> {
+    let mut stmt = tx.prepare("SELECT sequence,entity,record_id,payload,updated_at,deleted,version FROM changes WHERE user_id=? AND sequence>? ORDER BY sequence ASC LIMIT ?")?;
+    let mut rows = stmt.query(params![user, after, 501])?;
+    let mut result = Vec::new();
+    let mut cursor = after;
+    let mut more = false;
+    while let Some(r) = rows.next()? {
+        if result.len() == 500 {
+            more = true;
+            break;
+        }
+        let sequence: i64 = r.get(0)?;
+        let payload: Option<Vec<u8>> = match r.get_ref(3)? {
+            rusqlite::types::ValueRef::Null => None,
+            rusqlite::types::ValueRef::Text(bytes) | rusqlite::types::ValueRef::Blob(bytes) => {
+                Some(bytes.to_vec())
+            }
+            _ => return Err("invalid database payload".into()),
+        };
+        let time: String = r.get(4)?;
+        let version: i64 = r.get(6)?;
+        let mut item = json!({"entity": r.get::<_, String>(1)?, "id": r.get::<_, String>(2)?, "updatedAt": timestamp(parse_time(&time)?), "deleted": r.get::<_, bool>(5)?});
+        if sequence != 0 {
+            item["sequence"] = json!(sequence);
+        }
+        if version != 0 {
+            item["version"] = json!(version);
+        }
+        let mut encoded = serde_json::to_vec(&item)?;
+        if let Some(p) = payload {
+            serde_json::from_slice::<Box<RawValue>>(&crate::go_json::normalize(&p))?;
+            encoded.pop();
+            encoded.extend_from_slice(b",\"payload\":");
+            encoded.extend_from_slice(&crate::go_json::compact_raw(&p));
+            encoded.push(b'}');
+        }
+        result.push(encoded);
+        cursor = sequence;
+    }
+    Ok((result, cursor, more))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn db() -> Connection {
+        let db = crate::database::open(":memory:").unwrap();
+        db.execute(
+            "INSERT INTO users VALUES('u','u@example.com','hash','now')",
+            [],
+        )
+        .unwrap();
+        db.execute(
+            "INSERT INTO users VALUES('v','v@example.com','hash','now')",
+            [],
+        )
+        .unwrap();
+        db
+    }
+    fn change(id: &str, time: &str) -> Value {
+        json!({"entity":"category","id":id,"updatedAt":time,"payload":{"name":"Food"}})
+    }
+    fn sync(db: &mut Connection, body: Value) -> Result<Value, SyncError> {
+        synchronize(db, "u", &serde_json::to_vec(&body).unwrap())
+    }
+
+    #[test]
+    fn last_write_wins_and_replay_does_not_advance_sequence() {
+        let mut db = db();
+        let original = change("c", "2025-01-01T01:00:00+01:00");
+        let first = sync(&mut db, json!({"changes":[original.clone()]})).unwrap();
+        assert_eq!(first["cursor"], 1);
+        assert_eq!(first["changes"][0]["updatedAt"], "2025-01-01T00:00:00Z");
+        for c in [original, change("c", "2024-01-01T00:00:00Z")] {
+            let r = sync(&mut db, json!({"cursor":1,"changes":[c]})).unwrap();
+            assert_eq!(r["cursor"], 1);
+            assert_eq!(r["changes"], json!([]));
+        }
+        let deleted = json!({"entity":"category","id":"c","updatedAt":"2025-02-01T00:00:00.123400000Z","deleted":true,"payload":{"ignored":true}});
+        let r = sync(&mut db, json!({"cursor":1,"changes":[deleted]})).unwrap();
+        assert_eq!(r["changes"][0]["version"], 2);
+        assert_eq!(r["changes"][0]["deleted"], true);
+        assert!(r["changes"][0].get("payload").is_none());
+        assert_eq!(r["changes"][0]["updatedAt"], "2025-02-01T00:00:00.1234Z");
+    }
+
+    #[test]
+    fn pagination_and_user_isolation_preserve_global_sequence() {
+        let mut db = db();
+        let changes: Vec<Value> = (0..501)
+            .map(|i| change(&i.to_string(), "2025-01-01T00:00:00Z"))
+            .collect();
+        let r = sync(&mut db, json!({"changes":changes,"pull":false,"cursor":7})).unwrap();
+        assert_eq!(r["cursor"], 7);
+        assert_eq!(r["changes"], json!([]));
+        let r = sync(&mut db, json!({})).unwrap();
+        assert_eq!(r["changes"].as_array().unwrap().len(), 500);
+        assert_eq!(r["cursor"], 500);
+        assert_eq!(r["hasMore"], true);
+        let r = sync(&mut db, json!({"cursor":500})).unwrap();
+        assert_eq!(r["changes"].as_array().unwrap().len(), 1);
+        assert_eq!(r["cursor"], 501);
+        assert_eq!(r["hasMore"], false);
+        let r = synchronize(&mut db, "v", b"{}").unwrap();
+        assert_eq!(r["changes"], json!([]));
+        assert_eq!(r["cursor"], 0);
+    }
+
+    #[test]
+    fn all_pushes_roll_back_on_apply_or_pull_failure() {
+        let mut db = db();
+        db.execute_batch("CREATE TRIGGER reject_bad BEFORE INSERT ON records WHEN NEW.record_id='bad' BEGIN SELECT RAISE(ABORT, 'reject'); END;").unwrap();
+        let r = sync(
+            &mut db,
+            json!({"changes":[change("good","2025-01-01T00:00:00Z"),change("bad","2025-01-01T00:00:00Z")]}),
+        );
+        assert_eq!(r.unwrap_err().2, "Could not apply changes");
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM records", [], |r| r.get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+        db.execute("INSERT INTO changes(user_id,entity,record_id,updated_at,deleted,version) VALUES('u','category','broken','invalid-time',1,1)", []).unwrap();
+        assert_eq!(
+            sync(
+                &mut db,
+                json!({"changes":[change("good","2025-01-01T00:00:00Z")]})
+            )
+            .unwrap_err()
+            .2,
+            "Could not load changes"
+        );
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM records", [], |r| r.get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+    }
+
+    #[test]
+    fn strict_fields_and_go_null_defaults_and_validation() {
+        let mut db = db();
+        for raw in [
+            b"null".as_slice(),
+            b"{\"cursor\":null,\"deviceId\":null,\"changes\":null,\"pull\":null}",
+            b"{} {}",
+        ] {
+            assert!(synchronize(&mut db, "u", raw).is_ok());
+        }
+        for raw in [
+            b"{\"extra\":1}".as_slice(),
+            b"{\"changes\":[{\"extra\":1}]}",
+            b"{\"cursor\":1.0}",
+            b"{\"changes\":{}}",
+        ] {
+            assert_eq!(
+                synchronize(&mut db, "u", raw).unwrap_err().1,
+                "invalid_json"
+            );
+        }
+        let raw = br#"{"CHANGES":[{"ENTITY":"category","ID":"c","updatedAt":"2025-01-01T00:00:00Z","payload":null}]}"#;
+        let r = synchronize(&mut db, "u", raw).unwrap();
+        assert!(r["changes"][0].get("payload").unwrap().is_null());
+        assert_eq!(
+            sync(&mut db, json!({"cursor":-1})).unwrap_err().1,
+            "invalid_sync_request"
+        );
+        assert_eq!(
+            sync(&mut db, json!({"changes":[null]})).unwrap_err().2,
+            "entity must be transaction, category, or budget"
+        );
+        let mut c = change("c", "0001-01-01T00:00:00Z");
+        assert_eq!(
+            sync(&mut db, json!({"changes":[c.clone()]})).unwrap_err().2,
+            "updatedAt is required"
+        );
+        c["updatedAt"] = json!(timestamp(Utc::now() + Duration::hours(25)));
+        assert_eq!(
+            sync(&mut db, json!({"changes":[c]})).unwrap_err().2,
+            "updatedAt is too far in the future"
+        );
+        let mut c = change("c", "2025-01-01T00:00:00Z");
+        c["payload"] = json!("x".repeat(256 * 1024));
+        assert_eq!(
+            sync(&mut db, json!({"changes":[c]})).unwrap_err().2,
+            "payload is too large"
+        );
+    }
+    #[test]
+    fn go_timestamp_parser_accepts_fallbacks_and_rejects_chrono_extensions() {
+        for (input, expected) in [
+            ("2025-01-02T3:04:05Z", "2025-01-02T03:04:05Z"),
+            (
+                "2025-01-02T03:04:05,123456789123Z",
+                "2025-01-02T03:04:05.123456789Z",
+            ),
+            ("2025-01-02T03:04:05+24:60", "2025-01-01T02:04:05Z"),
+            ("0000-01-01T00:00:00Z", "0000-01-01T00:00:00Z"),
+        ] {
+            assert_eq!(timestamp(parse_time(input).unwrap()), expected, "{input}");
+        }
+        for input in [
+            "2025-01-02t03:04:05Z",
+            "2025-01-02T03:04:05z",
+            "2025-01-02T03:04:60Z",
+            "2025-01-02 03:04:05Z",
+            "2025-01-02T03:04:05+25:00",
+            "2025-01-02T03:04:05+01:61",
+            "2025-01-02T03:04:05.Z",
+            "2025-02-29T00:00:00Z",
+        ] {
+            assert!(parse_time(input).is_err(), "{input}");
+        }
+        let mut db = db();
+        let raw = br#"{"changes":[{"entity":"category","id":"c","updatedAt":"\u0032025-01-01T00:00:00Z","payload":null}]}"#;
+        assert_eq!(
+            synchronize(&mut db, "u", raw).unwrap_err().1,
+            "invalid_json"
+        );
+    }
+
+    #[test]
+    fn duplicate_fields_reuse_go_slice_elements_and_null_keeps_scalars() {
+        let mut db = db();
+        let raw = br#"{"changes":[{"entity":"category","id":"first","updatedAt":"2025-01-01T00:00:00Z","payload":null}],"changes":[{"id":"second","entity":null,"updatedAt":null}]}"#;
+        let result = synchronize(&mut db, "u", raw).unwrap();
+        assert_eq!(result["changes"][0]["id"], "second");
+        assert_eq!(result["changes"][0]["entity"], "category");
+        let raw = r#"{"curſor":1,"CHANGES":[{"entity":"category","id":"third","updatedAt":"2025-01-02T00:00:00Z","payload":true}]}"#;
+        let result = synchronize(&mut db, "u", raw.as_bytes()).unwrap();
+        assert_eq!(result["changes"][0]["id"], "third");
+        assert_eq!(result["changes"][0]["payload"], true);
+    }
+
+    #[test]
+    fn exact_payload_numbers_and_database_raw_text_are_preserved() {
+        let mut db = db();
+        let payload = r#"{ "large": 123456789012345678901234567890, "decimal": 0.123456789012345678901234567890, "exponent": 1e999 }"#;
+        let body = format!(
+            r#"{{"changes":[{{"entity":"transaction","id":"t","updatedAt":"2025-01-01T00:00:00Z","payload":{payload}}}]}}"#
+        );
+        let result = synchronize(&mut db, "u", body.as_bytes()).unwrap();
+        let stored: String = db
+            .query_row("SELECT payload FROM records", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(stored, payload);
+        let serialized = serde_json::to_string(&result["changes"][0]["payload"]).unwrap();
+        assert!(serialized.contains("123456789012345678901234567890"));
+        assert!(serialized.contains("0.123456789012345678901234567890"));
+        assert!(serialized.to_lowercase().contains("1e+999") || serialized.contains("1e999"));
+    }
+
+    #[test]
+    fn validation_limits_use_bytes_and_validate_whole_batch_before_writing() {
+        let mut db = db();
+        let mut invalid = change(&"я".repeat(65), "2025-01-01T00:00:00Z");
+        assert_eq!(
+            sync(
+                &mut db,
+                json!({"changes":[change("good","2025-01-01T00:00:00Z"),invalid.clone()]})
+            )
+            .unwrap_err()
+            .2,
+            "record id is invalid"
+        );
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM records", [], |r| r.get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+        invalid["id"] = json!("я".repeat(64));
+        assert!(sync(&mut db, json!({"changes":[invalid]})).is_ok());
+        let changes: Vec<_> = (0..5001)
+            .map(|_| change("c", "2025-01-01T00:00:00Z"))
+            .collect();
+        assert_eq!(
+            sync(&mut db, json!({"changes":changes})).unwrap_err().1,
+            "invalid_sync_request"
+        );
+        let raw = format!(
+            r#"{{"changes":[{{"entity":"category","id":"large","updatedAt":"2025-01-01T00:00:00Z","payload":{{ {} }} }}]}}"#,
+            " ".repeat(256 * 1024)
+        );
+        assert_eq!(
+            synchronize(&mut db, "u", raw.as_bytes()).unwrap_err().2,
+            "payload is too large"
+        );
+        let mut trailing = b"{} ".to_vec();
+        trailing.resize((2 << 20) + 100, b' ');
+        assert!(synchronize(&mut db, "u", &trailing).is_ok());
+    }
+
+    #[test]
+    fn wire_preserves_raw_payload_duplicates_unicode_escapes_and_invalid_utf8() {
+        let mut db = db();
+        let payload = b"{ \"duplicate\":1, \"duplicate\":2, \"surrogate\":\"\\ud800\", \"invalid\":\"\xff\", \"html\":\"<&>\" }";
+        let mut body = br#"{"changes":[{"entity":"category","id":"\ud800","updatedAt":"2025-01-01T00:00:00Z","payload":"#.to_vec();
+        body.extend_from_slice(payload);
+        body.extend_from_slice(b"}]}");
+        let response = synchronize_wire(&mut db, "u", &body).unwrap();
+        let id = "\"id\":\"�\"".as_bytes();
+        assert!(response.windows(id.len()).any(|w| w == id));
+        let expected = b"\"payload\":{\"duplicate\":1,\"duplicate\":2,\"surrogate\":\"\\ud800\",\"invalid\":\"\xff\",\"html\":\"\\u003c\\u0026\\u003e\"}";
+        assert!(response.windows(expected.len()).any(|w| w == expected));
+        let stored: Vec<u8> = db
+            .query_row("SELECT CAST(payload AS BLOB) FROM records", [], |r| {
+                r.get(0)
+            })
+            .unwrap();
+        assert_eq!(stored, payload);
+        let storage_type: String = db
+            .query_row("SELECT typeof(payload) FROM records", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(storage_type, "text");
+        let pulled = synchronize_wire(&mut db, "u", b"{}").unwrap();
+        assert!(pulled.windows(expected.len()).any(|w| w == expected));
+    }
+
+    #[test]
+    fn raw_payload_limits_count_original_bytes_before_unicode_replacement() {
+        let mut db = db();
+        let mut body = br#"{"changes":[{"entity":"category","id":"c","updatedAt":"2025-01-01T00:00:00Z","payload":""#.to_vec();
+        body.extend(std::iter::repeat_n(0xff, 256 * 1024 - 2));
+        body.extend_from_slice(b"\"}]}");
+        assert!(synchronize_wire(&mut db, "u", &body).is_ok());
+        let body = b"{\"changes\":[{\"entity\":\"category\",\"id\":\"\xff\",\"updatedAt\":\"2025-01-02T00:00:00Z\",\"payload\":null}]}";
+        assert!(synchronize_wire(&mut db, "u", body).is_ok());
+    }
+
+    #[test]
+    fn tombstone_resurrection_and_sequential_batch_conflicts_keep_history() {
+        let mut db = db();
+        let initial = change("c", "2025-01-01T00:00:00Z");
+        let tombstone =
+            json!({"entity":"category","id":"c","updatedAt":"2025-01-02T00:00:00Z","deleted":true});
+        let newer = change("c", "2025-01-03T00:00:00Z");
+        let result = sync(
+            &mut db,
+            json!({"changes":[initial.clone(),tombstone.clone(),initial,newer.clone(),newer]}),
+        )
+        .unwrap();
+        let changes = result["changes"].as_array().unwrap();
+        assert_eq!(changes.len(), 3);
+        assert_eq!(changes[0]["version"], 1);
+        assert_eq!(changes[1]["version"], 2);
+        assert_eq!(changes[2]["version"], 3);
+        assert_eq!(changes[2]["deleted"], false);
+        assert!(changes[1].get("payload").is_none());
+        let isolated = synchronize(
+            &mut db,
+            "v",
+            &serde_json::to_vec(&json!({"changes":[tombstone]})).unwrap(),
+        )
+        .unwrap();
+        assert_eq!(isolated["cursor"], 4);
+        let user = sync(&mut db, json!({"cursor":3})).unwrap();
+        assert_eq!(user["cursor"], 3);
+        assert_eq!(user["changes"], json!([]));
+    }
+
+    #[test]
+    fn concurrent_device_retries_have_one_winner_and_unique_sequences() {
+        use std::sync::{Arc, Barrier, Mutex};
+        let db = Arc::new(Mutex::new(db()));
+        let barrier = Arc::new(Barrier::new(8));
+        let mut handles = Vec::new();
+        for index in 0..8 {
+            let db = Arc::clone(&db);
+            let barrier = Arc::clone(&barrier);
+            handles.push(std::thread::spawn(move || {
+                barrier.wait();
+                let c = change("shared", &format!("2025-01-01T00:00:0{index}Z"));
+                let mut db = db.lock().unwrap();
+                sync(&mut db, json!({"changes":[c.clone(), c],"pull":false})).unwrap();
+            }));
+        }
+        for handle in handles {
+            handle.join().unwrap();
+        }
+        let mut db = db.lock().unwrap();
+        let (time, version): (String, i64) = db
+            .query_row("SELECT updated_at,version FROM records", [], |r| {
+                Ok((r.get(0)?, r.get(1)?))
+            })
+            .unwrap();
+        assert_eq!(time, "2025-01-01T00:00:07Z");
+        let result = sync(&mut db, json!({})).unwrap();
+        let changes = result["changes"].as_array().unwrap();
+        assert_eq!(changes.len() as i64, version);
+        for (index, change) in changes.iter().enumerate() {
+            assert_eq!(change["sequence"], (index + 1) as i64);
+            assert_eq!(change["version"], (index + 1) as i64);
+        }
+    }
+}
@@ -0,0 +1,408 @@
+#!/usr/bin/env python3
+"""Black-box Go/Rust compatibility and bilateral SQLite takeover checks (stdlib only)."""
+import argparse
+import base64
+import concurrent.futures
+import contextlib
+import http.client
+import hashlib
+import hmac
+import json
+import os
+from pathlib import Path
+import socket
+import sqlite3
+import subprocess
+import tempfile
+import time
+
+ROOT = Path(__file__).resolve().parents[1]
+SECRET = 'compatibility-test-secret-not-for-deployment-0123456789'
+PASSWORD = 'compatibility password'
+
+
+class Server:
+    def __init__(self, binary, database, **extra):
+        with socket.socket() as sock:
+            sock.bind(('127.0.0.1', 0))
+            self.port = sock.getsockname()[1]
+        env = os.environ.copy()
+        env.update(BALANCE_ADDR=f'127.0.0.1:{self.port}', BALANCE_DB_PATH=str(database),
+                   BALANCE_JWT_SECRET=SECRET, BALANCE_ACCESS_TTL='15m',
+                   BALANCE_REFRESH_TTL='720h', BALANCE_ALLOW_REGISTRATION='true',
+                   BALANCE_CORS_ORIGINS='https://allowed.example')
+        env.update(extra)
+        self.log = tempfile.TemporaryFile()
+        self.process = subprocess.Popen([str(binary)], env=env, cwd=ROOT,
+                                        stdout=self.log, stderr=self.log)
+        deadline = time.monotonic() + 20
+        while time.monotonic() < deadline:
+            if self.process.poll() is not None:
+                self.log.seek(0)
+                raise AssertionError(f'{binary.name} exited: {self.log.read().decode()}')
+            try:
+                if self.request('GET', '/health')[0] == 200:
+                    return
+            except (OSError, http.client.HTTPException):
+                pass
+            time.sleep(.05)
+        self.close()
+        raise AssertionError(f'{binary.name} startup timeout')
+
+    def close(self):
+        if self.process.poll() is None:
+            self.process.terminate()
+            try:
+                self.process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait()
+        self.log.close()
+
+    def request(self, method, path, body=None, token=None, headers=None, raw=None, decode_json=True):
+        hdr = dict(headers or {})
+        if token:
+            hdr['Authorization'] = f'Bearer {token}'
+        if body is not None:
+            raw = json.dumps(body).encode()
+            hdr['Content-Type'] = 'application/json'
+        connection = http.client.HTTPConnection('127.0.0.1', self.port, timeout=30)
+        try:
+            connection.request(method, path, body=raw, headers=hdr)
+            response = connection.getresponse()
+            data = response.read()
+            values = response.getheaders()
+            content_type = response.getheader('Content-Type', '')
+            if decode_json and data and 'application/json' in content_type:
+                data = json.loads(data)
+            return response.status, values, data
+        finally:
+            connection.close()
+
+
+@contextlib.contextmanager
+def running(binary, database, **extra):
+    server = Server(binary, database, **extra)
+    try:
+        yield server
+    finally:
+        server.close()
+
+
+def expect(result, status):
+    assert result[0] == status, f'expected HTTP {status}, got {result[0]}: {result[2]!r}'
+    return result[2]
+
+
+def credentials(email='User@Example.COM'):
+    return {'email': email, 'password': PASSWORD}
+
+
+def change(identifier, stamp='2025-01-01T00:00:00Z', **extra):
+    result = dict(entity='transaction', id=identifier, updatedAt=stamp,
+                  payload={'amount': 125.5, 'note': 'Тест €', 'nested': [None, True, 1]})
+    result.update(extra)
+    return result
+
+
+def sync(server, token, changes=(), cursor=0, **extra):
+    return expect(server.request('POST', '/v1/sync',
+        dict(cursor=cursor, deviceId='compatibility', changes=list(changes), **extra), token), 200)
+
+
+def scenario(binary, database):
+    """Return deterministic observable results; assertions check generated sessions too."""
+    observed = {}
+    with running(binary, database) as server:
+        for method, path, body in [
+            ('GET', '/v1/me', None), ('GET', '/v1/missing', None),
+            ('POST', '/v1/auth/register', {'email': 'bad', 'password': PASSWORD}),
+            ('POST', '/v1/auth/register', {'email': 'a@b', 'password': 'short'}),
+            ('POST', '/v1/auth/login', credentials()),
+            ('POST', '/v1/auth/refresh', {}),
+            ('POST', '/v1/auth/refresh', {'refreshToken': 'bad'}),
+            ('POST', '/v1/auth/login', {'unknown': 1}),
+        ]:
+            result = server.request(method, path, body)
+            observed[f'{method} {path} {body}'] = (result[0], result[2])
+        for path in ['/','/app.js','/styles.css','/icon.svg','/manifest.webmanifest',
+                     '/service-worker.js','/missing-file']:
+            result = server.request('GET', path)
+            head = {k.lower(): v for k, v in result[1]}
+            keys = ['content-type','cache-control','x-content-type-options',
+                    'x-frame-options','referrer-policy','permissions-policy','content-security-policy']
+            observed[path] = (result[0], {k: head.get(k) for k in keys}, result[2])
+            assert server.request('HEAD', path)[2] == b''
+        for origin in ['https://allowed.example', 'https://denied.example']:
+            result = server.request('OPTIONS', '/v1/sync', headers={'Origin': origin})
+            observed[origin] = (result[0], {k.lower(): v for k, v in result[1]
+                                           if k.lower().startswith('access-control') or k.lower() == 'vary'})
+        session_result = server.request('POST', '/v1/auth/register', credentials(),
+                                        headers={'X-Forwarded-Proto': 'https'})
+        session = expect(session_result, 201)
+        assert session['user']['email'] == 'user@example.com'
+        cookies = [v for k, v in session_result[1] if k.lower() == 'set-cookie']
+        assert len(cookies) == 2
+        for cookie in cookies:
+            assert all(part in cookie for part in ['HttpOnly','Secure','SameSite=Strict','Path=/'])
+        cookie_header = '; '.join(c.split(';')[0] for c in cookies)
+        assert expect(server.request('GET', '/v1/me', headers={'Cookie': cookie_header}), 200) == session['user']
+        token = session['accessToken']
+        assert expect(server.request('GET', '/v1/me', token=token), 200) == session['user']
+        observed['invalid bearer'] = expect(server.request('GET', '/v1/me', token='bad'), 401)
+        observed['duplicate'] = expect(server.request('POST', '/v1/auth/register', credentials()), 409)
+        expect(server.request('POST', '/v1/auth/login', credentials()), 200)
+        for payload in [{'cursor': -1}, {'extra': 1}, {'changes': [change('x', entity='unknown')]},
+                        {'changes': [change('')]}, {'changes': [dict(entity='category', id='x')]},
+                        {'changes': [change('x', payload=None)]}]:
+            result = server.request('POST', '/v1/sync', payload, token)
+            data = result[2]
+            if isinstance(data, dict):
+                data.pop('serverTime', None)
+            observed[f'invalid sync {payload}'] = (result[0], data)
+        observed['malformed'] = expect(server.request('POST', '/v1/sync', token=token, raw=b'{'), 400)
+        first = sync(server, token, [change('one'), change('cat', entity='category'), change('budget', entity='budget')])
+        first.pop('serverTime')
+        observed['first'] = first
+        repeated = sync(server, token, [change('one'), change('one', '2024-01-01T00:00:00Z')], cursor=first['cursor'])
+        repeated.pop('serverTime')
+        observed['idempotent'] = repeated
+        assert repeated['changes'] == [] and repeated['cursor'] == first['cursor']
+        deleted = sync(server, token, [change('one', '2025-01-02T00:00:00Z', deleted=True)], cursor=first['cursor'])
+        deleted.pop('serverTime')
+        observed['deleted'] = deleted
+        assert deleted['changes'][0]['deleted'] and 'payload' not in deleted['changes'][0]
+        pushed = sync(server, token, [change(f'page-{n}') for n in range(505)], cursor=deleted['cursor'], pull=False)
+        assert pushed['cursor'] == deleted['cursor'] and pushed['changes'] == []
+        page = sync(server, token, cursor=deleted['cursor'])
+        assert len(page['changes']) == 500 and page['hasMore']
+        tail = sync(server, token, cursor=page['cursor'])
+        assert len(tail['changes']) == 5 and not tail['hasMore']
+        observed['pagination'] = (page['cursor'], tail['cursor'])
+        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+            results = list(pool.map(lambda n: sync(server, token, [change(f'parallel-{n}')], pull=False), range(16)))
+        assert all(result['changes'] == [] for result in results)
+        parallel = sync(server, token, cursor=tail['cursor'])
+        assert len(parallel['changes']) == 16
+        assert len({x['sequence'] for x in parallel['changes']}) == 16
+        other = expect(server.request('POST', '/v1/auth/register', credentials('other@example.com')), 201)
+        assert sync(server, other['accessToken'])['changes'] == []
+        refreshed = expect(server.request('POST', '/v1/auth/refresh', {}, headers={'Cookie': cookie_header}), 200)
+        assert refreshed['refreshToken'] != session['refreshToken']
+        observed['refresh replay'] = expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': session['refreshToken']}), 401)
+        logged_out = server.request('POST', '/v1/auth/logout', {'refreshToken': refreshed['refreshToken']})
+        expect(logged_out, 204)
+        assert len([v for k,v in logged_out[1] if k.lower() == 'set-cookie' and 'Max-Age=0' in v]) == 2
+        observed['logout refresh'] = expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': refreshed['refreshToken']}), 401)
+    # Fresh limiter state permits exactly 30 authentication attempts, including invalid JSON.
+    with running(binary, database) as server:
+        statuses = [server.request('POST', '/v1/auth/login', raw=b'{')[0] for _ in range(31)]
+        assert statuses == [400] * 30 + [429], statuses
+        result = server.request('POST', '/v1/auth/login', raw=b'{')
+        assert {k.lower():v for k,v in result[1]}.get('retry-after') == '60'
+        observed['rate limit'] = result[2]
+    with running(binary, database, BALANCE_ALLOW_REGISTRATION='false', BALANCE_ACCESS_TTL='1s') as server:
+        observed['registration disabled'] = expect(server.request('POST', '/v1/auth/register', credentials()), 403)
+        session = expect(server.request('POST', '/v1/auth/login', credentials()), 200)
+        time.sleep(1.15)
+        observed['expired access'] = expect(server.request('GET', '/v1/me', token=session['accessToken']), 401)
+    return observed
+
+
+def takeover(first, second, database):
+    with running(first, database) as server:
+        session = expect(server.request('POST', '/v1/auth/register', credentials()), 201)
+        original = sync(server, session['accessToken'], [change('persisted')])
+    with running(second, database) as server:
+        assert expect(server.request('GET', '/v1/me', token=session['accessToken']), 200) == session['user']
+        assert expect(server.request('POST', '/v1/auth/login', credentials()), 200)['user'] == session['user']
+        assert sync(server, session['accessToken'])['changes'] == original['changes']
+        next_session = expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': session['refreshToken']}), 200)
+        updated = sync(server, next_session['accessToken'], [change('persisted','2025-01-02T00:00:00Z',deleted=True)], cursor=original['cursor'])
+        assert len(updated['changes']) == 1 and updated['changes'][0]['version'] == 2
+    with running(first, database) as server:
+        assert expect(server.request('GET', '/v1/me', token=next_session['accessToken']), 200) == session['user']
+        expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': session['refreshToken']}), 401)
+        expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': next_session['refreshToken']}), 200)
+        restored = sync(server, session['accessToken'], cursor=original['cursor'])
+        assert restored['changes'] == updated['changes'] and restored['cursor'] == updated['cursor']
+        expect(server.request('POST', '/v1/auth/login', credentials()), 200)
+
+
+def edge_scenarios(binary, database):
+    """Exercise decoder, routing, validation, range and session boundary behavior."""
+    observed = {}
+    def record(name, result, headers=()):
+        selected = {k.lower(): v for k, v in result[1] if k.lower() in headers}
+        data = result[2]
+        if isinstance(data, dict):
+            data = {k: v for k, v in data.items() if k != 'serverTime'}
+        observed[name] = (result[0], selected, data)
+
+    with running(binary, database) as server:
+        session = expect(server.request('POST', '/v1/auth/register', credentials()), 201)
+        token = session['accessToken']
+        for method in ['POST', 'PUT', 'DELETE', 'PATCH', 'HEAD']:
+            for path in ['/health', '/v1/me', '/v1/sync', '/app.js', '/v1/auth/login']:
+                record((method, path), server.request(method, path), ('allow', 'content-type'))
+        for path in ['/index.html', '/a/../app.js', '//app.js', '/app.js/',
+                     '/v1', '/v1/', '/v1//me', '/v1/../app.js', '/%61pp.js', '/%2e%2e/app.js']:
+            record(('path', path), server.request('GET', path), ('location', 'content-type', 'cache-control'))
+        for range_value in ['bytes=0-9', 'bytes=-10', 'bytes=10-', 'bytes=99999999-',
+                            'bytes=bad', 'items=0-1', 'bytes=10-1', 'bytes=0-0,2-2']:
+            result = server.request('GET', '/icon.svg', headers={'Range': range_value})
+            if result[0] == 206 and 'multipart/' in {k.lower():v for k,v in result[1]}.get('content-type', ''):
+                # Multipart boundaries are randomly generated in the reference.
+                assert b'Content-Range: bytes 0-0/' in result[2]
+                assert b'Content-Range: bytes 2-2/' in result[2]
+                observed[('range', range_value)] = result[0]
+            else:
+                record(('range', range_value), result, ('content-range', 'accept-ranges', 'content-type'))
+        raw_sync = [b'null', b'{} trailing', b'{"CURSOR":null,"CHANGES":null}',
+                    b'{"cursor":0,"cursor":null}', b'{"cursor":1.0}', b'[]',
+                    b'{"changes":[null]}', b'{"deviceId":false}', b'{"pull":0}',
+                    b'{"changes":[],"changes":null}', b'{"cursor":9223372036854775808}']
+        for n, raw in enumerate(raw_sync):
+            record(('raw sync', n), server.request('POST', '/v1/sync', token=token, raw=raw))
+        cases = [change('x', entity='Category'), change('x'*129),
+                 change('x', payload=None), change('x', payload=[1,2,3]),
+                 change('x', stamp='0001-01-01T00:00:00Z'),
+                 change('x', stamp='2025-01-01t00:00:00z'),
+                 change('x', stamp='2025-01-01T0:00:00Z'),
+                 change('x', stamp='2025-01-01T00:00:60Z'),
+                 change('x', stamp='2025-01-01T00:00:00,123456789123Z'),
+                 change('x', stamp='2025-01-01T00:00:00+24:00'),
+                 change('x', stamp='2025-01-01T00:00:00+00:60'),
+                 change('x', stamp='2025-01-01T00:00:00.123456789123Z'),
+                 change('x', stamp='9999-01-01T00:00:00Z'),
+                 change('x', payload='x'*(256*1024))]
+        for n, value in enumerate(cases):
+            record(('sync validation', n), server.request('POST', '/v1/sync', {'changes':[value]}, token))
+        for n, raw in enumerate([b'null', b'{} tail', b'{"refreshToken":null}', b'{"REFRESHTOKEN":"bad"}', b'{"refreshToken":1}', b'{"extra":1}']):
+            record(('refresh JSON', n), server.request('POST', '/v1/auth/refresh', raw=raw))
+            record(('logout JSON', n), server.request('POST', '/v1/auth/logout', raw=raw))
+        for auth in ['', 'Basic bad', 'bearer bad', 'Bearer ', 'Bearer bad']:
+            record(('bearer', auth), server.request('GET', '/v1/me', headers={'Authorization':auth}))
+        # Verify signed JWTs with claims from Go's permissive JSON decoder.
+        for n, claims in enumerate([
+            {'sub':'u','email':'e','iss':'balance-server','exp':4102444800},
+            {'SUB':'u','EMAIL':'e','ISS':'balance-server','EXP':4102444800},
+            {'sub':'u','email':None,'iss':'balance-server','exp':4102444800},
+            {'sub':'u','iss':'balance-server','exp':4102444800,'unknown':123},
+            {'sub':'u','iss':'wrong','exp':4102444800},
+            {'sub':'u','iss':'balance-server','exp':1},
+            {'sub':'','iss':'balance-server','exp':4102444800},
+        ]):
+            encode = lambda data: base64.urlsafe_b64encode(data).rstrip(b'=')
+            unsigned = b'ignored-header.' + encode(json.dumps(claims).encode())
+            signed = (unsigned+b'.'+encode(hmac.new(SECRET.encode(), unsigned, hashlib.sha256).digest())).decode()
+            record(('signed claims', n), server.request('GET', '/v1/me', token=signed))
+        # Only one simultaneous rotation of a given refresh session may succeed.
+        fresh = expect(server.request('POST','/v1/auth/login',credentials()),200)
+        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+            statuses = sorted(pool.map(lambda _: server.request('POST','/v1/auth/refresh',{'refreshToken':fresh['refreshToken']})[0], range(4)))
+        assert statuses == [200,401,401,401], statuses
+    with running(binary, database) as server:
+        for n, raw in enumerate([b'null', b'{"email":null,"password":null}',
+                                b'{"EMAIL":"bad","PASSWORD":"abcdefgh"}',
+                                b'{"email":"bad","email":null}', b'{"email":false}',
+                                b'{"email":"bad"} trailing', b'{"email":"bad","unknown":1}']):
+            for endpoint in ['register','login']:
+                record((endpoint, n), server.request('POST', '/v1/auth/'+endpoint, raw=raw))
+        for n, value in enumerate([credentials('a@b'),dict(email='utf8@example.com',password='я'*4),
+                                  dict(email='long@example.com',password='x'*1025)]):
+            result = server.request('POST','/v1/auth/register',value)
+            if result[0] == 201:
+                assert expect(server.request('POST','/v1/auth/login',value),200)['user'] == result[2]['user']
+                observed[('password boundary',n)] = result[0]
+            else:
+                record(('password boundary',n), result)
+    with running(binary,database,BALANCE_REFRESH_TTL='1ms') as server:
+        session = expect(server.request('POST','/v1/auth/login',credentials()),200)
+        time.sleep(.05)
+        record('expired refresh',server.request('POST','/v1/auth/refresh',{'refreshToken':session['refreshToken']}))
+    return observed
+
+
+def raw_json_scenarios(binary, database):
+    """RawMessage must retain numbers, duplicate keys and unusual Unicode bytes."""
+    observed = {}
+    with running(binary, database) as server:
+        session = expect(server.request('POST','/v1/auth/register',credentials()),201)
+        token = session['accessToken']
+        cursor = 0
+        for n, payload in enumerate([
+            b'{ "duplicate":1, "duplicate":2, "n":123456789012345678901234567890, "big":1e999 }',
+            br'{"lone":"\ud800","low":"\udfff","pair":"\ud83d\ude00"}',
+            b'{"invalid":"\xff\xc0\xed\xa0\x80"}',
+            b'{"html":"<>&","separator":"\xe2\x80\xa8\xe2\x80\xa9"}',
+            br'{"nested":[{"payload":{"id":"inner"}},"quote\"end",null,true]}',
+        ]):
+            raw = (b'{"cursor":'+str(cursor).encode()+b',"changes":[{"entity":"transaction","id":"wire-'+str(n).encode()+
+                   b'","updatedAt":"2025-01-01T00:00:00Z","payload":'+payload+b'}]}')
+            result = server.request('POST','/v1/sync',token=token,raw=raw,decode_json=False)
+            assert result[0] == 200, result
+            data = result[2]
+            # Latin-1 preserves byte indices while Python locates the complete JSON value.
+            start = data.index(b'"payload":') + len(b'"payload":')
+            _, end = json.JSONDecoder().raw_decode(data[start:].decode('latin-1'))
+            observed[('wire payload',n)] = data[start:start+end]
+            decoded = json.loads(data.decode('utf-8','replace'))
+            cursor = decoded['cursor']
+            assert len(decoded['changes']) == 1
+        for n, raw in enumerate([br'{"email":"\ud800","password":"abcdefgh"}',
+                                 b'{"email":"\xff@example.com","password":"abcdefgh"}']):
+            # Login has deterministic invalid credentials and exercises scalar normalization.
+            result = server.request('POST','/v1/auth/login',raw=raw)
+            observed[('unicode credentials',n)] = (result[0],result[2])
+        for n, raw_id in enumerate([br'"\ud800"',b'"\xff"']):
+            raw = (b'{"cursor":'+str(cursor).encode()+b',"changes":[{"entity":"category","id":'+raw_id+
+                   b',"updatedAt":"2025-01-02T00:00:00Z","payload":{}}]}')
+            result = server.request('POST','/v1/sync',token=token,raw=raw)
+            assert result[0] == 200, result
+            data = result[2]
+            data.pop('serverTime')
+            observed[('unicode id',n)] = data
+            cursor = data['cursor']
+    with sqlite3.connect(database) as db:
+        db.text_factory = bytes
+        observed['raw database payloads'] = db.execute('SELECT record_id,payload FROM records ORDER BY record_id').fetchall()
+    return observed
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--go-binary', type=Path)
+    parser.add_argument('--rust-binary', type=Path, default=ROOT / 'target/debug/balance-server')
+    args = parser.parse_args()
+    with tempfile.TemporaryDirectory(prefix='balance-compatibility-') as directory:
+        directory = Path(directory)
+        go = args.go_binary.resolve() if args.go_binary else directory / 'go-server'
+        if args.go_binary is None:
+            subprocess.run(['go','build','-o',str(go),'./cmd/balance-server'], cwd=ROOT, check=True)
+        rust = args.rust_binary.resolve()
+        assert rust.is_file(), f'Build Rust first: cargo build --locked (missing {rust})'
+        print('Running Go reference scenarios...', flush=True)
+        expected = scenario(go, directory / 'go.db')
+        print('Running Rust scenarios...', flush=True)
+        actual = scenario(rust, directory / 'rust.db')
+        for key in expected:
+            assert actual[key] == expected[key], f'Differential mismatch: {key}\nGo: {expected[key]!r}\nRust: {actual[key]!r}'
+        print('Comparing HTTP/JSON edge cases...', flush=True)
+        expected = edge_scenarios(go, directory / 'go-edge.db')
+        actual = edge_scenarios(rust, directory / 'rust-edge.db')
+        for key in expected:
+            assert actual[key] == expected[key], f'Differential mismatch: {key}\nGo: {expected[key]!r}\nRust: {actual[key]!r}'
+        print('Comparing raw JSON payload and Unicode preservation...', flush=True)
+        expected = raw_json_scenarios(go, directory / 'go-raw.db')
+        actual = raw_json_scenarios(rust, directory / 'rust-raw.db')
+        for key in expected:
+            assert actual[key] == expected[key], f'Differential mismatch: {key}\nGo: {expected[key]!r}\nRust: {actual[key]!r}'
+        for first, second, name in [(go,rust,'go-rust'),(rust,go,'rust-go')]:
+            print(f'Checking {name} SQLite takeover and rollback...', flush=True)
+            takeover(first, second, directory / f'{name}.db')
+    print('PASS: HTTP differential scenarios and bilateral SQLite takeover/rollback')
+
+
+if __name__ == '__main__':
+    main()