diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..1dfcd0f --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,296 @@ +# This file was autogenerated by dist: https://axodotdev.github.io/cargo-dist +# +# Copyright 2022-2024, axodotdev +# SPDX-License-Identifier: MIT or Apache-2.0 +# +# CI that: +# +# * checks for a Git Tag that looks like a release +# * builds artifacts with dist (archives, installers, hashes) +# * uploads those artifacts to temporary workflow zip +# * on success, uploads the artifacts to a GitHub Release +# +# Note that the GitHub Release will be created with a generated +# title/body based on your changelogs. + +name: Release +permissions: + "contents": "write" + +# This task will run whenever you push a git tag that looks like a version +# like "1.0.0", "v0.1.0-prerelease.1", "my-app/0.1.0", "releases/v1.0.0", etc. +# Various formats will be parsed into a VERSION and an optional PACKAGE_NAME, where +# PACKAGE_NAME must be the name of a Cargo package in your workspace, and VERSION +# must be a Cargo-style SemVer Version (must have at least major.minor.patch). +# +# If PACKAGE_NAME is specified, then the announcement will be for that +# package (erroring out if it doesn't have the given version or isn't dist-able). +# +# If PACKAGE_NAME isn't specified, then the announcement will be for all +# (dist-able) packages in the workspace with that version (this mode is +# intended for workspaces with only one dist-able package, or with all dist-able +# packages versioned/released in lockstep). +# +# If you push multiple tags at once, separate instances of this workflow will +# spin up, creating an independent announcement for each one. However, GitHub +# will hard limit this to 3 tags per commit, as it will assume more tags is a +# mistake. +# +# If there's a prerelease-style suffix to the version, then the release(s) +# will be marked as a prerelease. +on: + pull_request: + push: + tags: + - '**[0-9]+.[0-9]+.[0-9]+*' + +jobs: + # Run 'dist plan' (or host) to determine what tasks we need to do + plan: + runs-on: "ubuntu-22.04" + outputs: + val: ${{ steps.plan.outputs.manifest }} + tag: ${{ !github.event.pull_request && github.ref_name || '' }} + tag-flag: ${{ !github.event.pull_request && format('--tag={0}', github.ref_name) || '' }} + publishing: ${{ !github.event.pull_request }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install dist + # we specify bash to get pipefail; it guards against the `curl` command + # failing. otherwise `sh` won't catch that `curl` returned non-0 + shell: bash + run: "curl --proto '=https' --tlsv1.2 -LsSf https://github.com/axodotdev/cargo-dist/releases/download/v0.32.0/cargo-dist-installer.sh | sh" + - name: Cache dist + uses: actions/upload-artifact@v7 + with: + name: cargo-dist-cache + path: ~/.cargo/bin/dist + # sure would be cool if github gave us proper conditionals... + # so here's a doubly-nested ternary-via-truthiness to try to provide the best possible + # functionality based on whether this is a pull_request, and whether it's from a fork. + # (PRs run on the *source* but secrets are usually on the *target* -- that's *good* + # but also really annoying to build CI around when it needs secrets to work right.) + - id: plan + run: | + dist ${{ (!github.event.pull_request && format('host --steps=create --tag={0}', github.ref_name)) || 'plan' }} --output-format=json > plan-dist-manifest.json + echo "dist ran successfully" + cat plan-dist-manifest.json + echo "manifest=$(jq -c "." plan-dist-manifest.json)" >> "$GITHUB_OUTPUT" + - name: "Upload dist-manifest.json" + uses: actions/upload-artifact@v7 + with: + name: artifacts-plan-dist-manifest + path: plan-dist-manifest.json + + # Build and packages all the platform-specific things + build-local-artifacts: + name: build-local-artifacts (${{ join(matrix.targets, ', ') }}) + # Let the initial task tell us to not run (currently very blunt) + needs: + - plan + if: ${{ fromJson(needs.plan.outputs.val).ci.github.artifacts_matrix.include != null && (needs.plan.outputs.publishing == 'true' || fromJson(needs.plan.outputs.val).ci.github.pr_run_mode == 'upload') }} + strategy: + fail-fast: false + # Target platforms/runners are computed by dist in create-release. + # Each member of the matrix has the following arguments: + # + # - runner: the github runner + # - dist-args: cli flags to pass to dist + # - install-dist: expression to run to install dist on the runner + # + # Typically there will be: + # - 1 "global" task that builds universal installers + # - N "local" tasks that build each platform's binaries and platform-specific installers + matrix: ${{ fromJson(needs.plan.outputs.val).ci.github.artifacts_matrix }} + runs-on: ${{ matrix.runner }} + container: ${{ matrix.container && matrix.container.image || null }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + BUILD_MANIFEST_NAME: target/distrib/${{ join(matrix.targets, '-') }}-dist-manifest.json + steps: + - name: enable windows longpaths + run: | + git config --global core.longpaths true + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install Rust non-interactively if not already installed + if: ${{ matrix.container }} + run: | + if ! command -v cargo > /dev/null 2>&1; then + curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y + echo "$HOME/.cargo/bin" >> $GITHUB_PATH + fi + - name: Install dist + run: ${{ matrix.install_dist.run }} + # Get the dist-manifest + - name: Fetch local artifacts + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: target/distrib/ + merge-multiple: true + - name: Install dependencies + run: | + ${{ matrix.packages_install }} + - name: Build artifacts + run: | + # Actually do builds and make zips and whatnot + dist build ${{ needs.plan.outputs.tag-flag }} --print=linkage --output-format=json ${{ matrix.dist_args }} > dist-manifest.json + echo "dist ran successfully" + - id: cargo-dist + name: Post-build + # We force bash here just because github makes it really hard to get values up + # to "real" actions without writing to env-vars, and writing to env-vars has + # inconsistent syntax between shell and powershell. + shell: bash + run: | + # Parse out what we just built and upload it to scratch storage + echo "paths<> "$GITHUB_OUTPUT" + dist print-upload-files-from-manifest --manifest dist-manifest.json >> "$GITHUB_OUTPUT" + echo "EOF" >> "$GITHUB_OUTPUT" + + cp dist-manifest.json "$BUILD_MANIFEST_NAME" + - name: "Upload artifacts" + uses: actions/upload-artifact@v7 + with: + name: artifacts-build-local-${{ join(matrix.targets, '_') }} + path: | + ${{ steps.cargo-dist.outputs.paths }} + ${{ env.BUILD_MANIFEST_NAME }} + + # Build and package all the platform-agnostic(ish) things + build-global-artifacts: + needs: + - plan + - build-local-artifacts + runs-on: "ubuntu-22.04" + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + BUILD_MANIFEST_NAME: target/distrib/global-dist-manifest.json + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install cached dist + uses: actions/download-artifact@v8 + with: + name: cargo-dist-cache + path: ~/.cargo/bin/ + - run: chmod +x ~/.cargo/bin/dist + # Get all the local artifacts for the global tasks to use (for e.g. checksums) + - name: Fetch local artifacts + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: target/distrib/ + merge-multiple: true + - id: cargo-dist + shell: bash + run: | + dist build ${{ needs.plan.outputs.tag-flag }} --output-format=json "--artifacts=global" > dist-manifest.json + echo "dist ran successfully" + + # Parse out what we just built and upload it to scratch storage + echo "paths<> "$GITHUB_OUTPUT" + jq --raw-output ".upload_files[]" dist-manifest.json >> "$GITHUB_OUTPUT" + echo "EOF" >> "$GITHUB_OUTPUT" + + cp dist-manifest.json "$BUILD_MANIFEST_NAME" + - name: "Upload artifacts" + uses: actions/upload-artifact@v7 + with: + name: artifacts-build-global + path: | + ${{ steps.cargo-dist.outputs.paths }} + ${{ env.BUILD_MANIFEST_NAME }} + # Determines if we should publish/announce + host: + needs: + - plan + - build-local-artifacts + - build-global-artifacts + # Only run if we're "publishing", and only if plan, local and global didn't fail (skipped is fine) + if: ${{ always() && needs.plan.result == 'success' && needs.plan.outputs.publishing == 'true' && (needs.build-global-artifacts.result == 'skipped' || needs.build-global-artifacts.result == 'success') && (needs.build-local-artifacts.result == 'skipped' || needs.build-local-artifacts.result == 'success') }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + runs-on: "ubuntu-22.04" + outputs: + val: ${{ steps.host.outputs.manifest }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install cached dist + uses: actions/download-artifact@v8 + with: + name: cargo-dist-cache + path: ~/.cargo/bin/ + - run: chmod +x ~/.cargo/bin/dist + # Fetch artifacts from scratch-storage + - name: Fetch artifacts + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: target/distrib/ + merge-multiple: true + - id: host + shell: bash + run: | + dist host ${{ needs.plan.outputs.tag-flag }} --steps=upload --steps=release --output-format=json > dist-manifest.json + echo "artifacts uploaded and released successfully" + cat dist-manifest.json + echo "manifest=$(jq -c "." dist-manifest.json)" >> "$GITHUB_OUTPUT" + - name: "Upload dist-manifest.json" + uses: actions/upload-artifact@v7 + with: + # Overwrite the previous copy + name: artifacts-dist-manifest + path: dist-manifest.json + # Create a GitHub Release while uploading all files to it + - name: "Download GitHub Artifacts" + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: artifacts + merge-multiple: true + - name: Cleanup + run: | + # Remove the granular manifests + rm -f artifacts/*-dist-manifest.json + - name: Create GitHub Release + env: + PRERELEASE_FLAG: "${{ fromJson(steps.host.outputs.manifest).announcement_is_prerelease && '--prerelease' || '' }}" + ANNOUNCEMENT_TITLE: "${{ fromJson(steps.host.outputs.manifest).announcement_title }}" + ANNOUNCEMENT_BODY: "${{ fromJson(steps.host.outputs.manifest).announcement_github_body }}" + RELEASE_COMMIT: "${{ github.sha }}" + run: | + # Write and read notes from a file to avoid quoting breaking things + echo "$ANNOUNCEMENT_BODY" > $RUNNER_TEMP/notes.txt + + gh release create "${{ needs.plan.outputs.tag }}" --target "$RELEASE_COMMIT" $PRERELEASE_FLAG --title "$ANNOUNCEMENT_TITLE" --notes-file "$RUNNER_TEMP/notes.txt" artifacts/* + + announce: + needs: + - plan + - host + # use "always() && ..." to allow us to wait for all publish jobs while + # still allowing individual publish jobs to skip themselves (for prereleases). + # "host" however must run to completion, no skipping allowed! + if: ${{ always() && needs.host.result == 'success' }} + runs-on: "ubuntu-22.04" + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive diff --git a/.gitignore b/.gitignore index 6312e72..a0e9bbd 100644 --- a/.gitignore +++ b/.gitignore @@ -4,5 +4,6 @@ *.toml !Cargo.toml !config.example.toml +!dist-workspace.toml .DS_Store *.log \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index dab9a6a..f6ae36d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ name = "flowsurface-server" version = "0.1.0" edition = "2024" license = "MIT" +repository = "https://github.com/akenshaw/fs-server" [dependencies] # HTTP / async runtime / concurrency @@ -45,4 +46,9 @@ tracing = { version = "0.1", default-features = false, features = ["std"] } tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "env-filter", "ansi"] } # Error handling -anyhow = "1" \ No newline at end of file +anyhow = "1" + +# The profile that 'dist' will build with +[profile.dist] +inherits = "release" +lto = "thin" diff --git a/README.md b/README.md index 18354cc..3674d02 100644 --- a/README.md +++ b/README.md @@ -1,52 +1,59 @@ # flowsurface-server -A crypto trade data collector and server. +A trade data collector for crypto markets, with an embedded database and REST API. -Connects to crypto exchange WebSocket streams via [flowsurface-exchange](https://crates.io/crates/flowsurface-exchange), -persists trades to an embedded [DuckDB](https://duckdb.org) database, and serves -them over a REST API with optional Arrow IPC export. +- Connects to exchange WebSocket streams via [flowsurface-exchange](https://crates.io/crates/flowsurface-exchange) +- Persists trades to [DuckDB](https://duckdb.org) +- Serves data via a REST API, as JSON or [Arrow IPC](https://arrow.apache.org/) stream formats ## Quick start +1. **Copy the template**: + ```bash -# 1. Copy the example config and edit to suit cp config.example.toml config.toml -# edit config.toml to set your exchange whitelist, base assets, etc. +``` + +> See the [basic settings](#basic) and edit `config.toml` + +2. **Run** + +```bash +# looks for `config.toml` next to the binary. +./flowsurface-server +``` + +Or to use a custom config path: -# 2. Run -./flowsurface-server # looks for config.toml next to binary or CWD +```bash ./flowsurface-server --config /path/to/config.toml ``` > If you run without a config file, the server will write the -> template to the given path and **exit with code 2** — this is deliberate -> so that systemd/supervisors can distinguish "not yet configured" from a -> crash. Simply edit the generated file and re-run. +> template to the given path and then exit. Simply edit the generated file and re-run. ## Configuration -Full reference — see [`config.example.toml`](config.example.toml) for all +See [`config.example.toml`](config.example.toml) for all available options with inline documentation. -| Situation | Behaviour | -| --------------------------------- | ------------------------------------------------------------------------ | -| `bind_address = "127.0.0.1:8080"` | Plain HTTP, no auth required | -| `bind_address = "0.0.0.0:8080"` | HTTPS (self-signed cert), auth token + cert fingerprint generated | -| `discovery_mode = true` (default) | Fetch metadata for **all** exchanges at startup to populate `/exchanges` | -| `discovery_mode = false` | Only fetch metadata for whitelisted venues | - -### Key settings - -| Option | Default | Description | -| ---------------------- | ---------------------- | --------------------------------------------------------------------------- | -| `bind_address` | — | Socket address to bind (e.g. `127.0.0.1:8080`) | -| `data_dir` | `"./data"` | Directory for DuckDB, auth token, TLS certs/keys | -| `base_assets` | — | Base assets expanded via whitelist templates (e.g. `["BTC"]`) | -| `flush_interval_ms` | `2000` | Batch flush interval (ms); lower = less data loss, higher = I/O efficient | -| `data_retention_hours` | `48` | Trades older than this are purged on startup & periodically | -| `discovery_mode` | `true` | Fetch metadata for all exchange variants so `/exchanges` is fully populated | -| `max_buffered_trades` | `200000` | Max trades in memory buffer before dropping (OOM guard) | -| `tls_domain` | `"flowsurface-server"` | Domain in the self-signed TLS cert's SAN | +### Basic + +| Option | Default | Description | +| ---------------------- | ------- | ------------------------------------------------------------------------------------------------------- | +| `bind_address` | — | Socket address to bind (`127.0.0.1:8080` = local-only plain HTTP; `0.0.0.0:8080` = remote HTTPS + auth) | +| `base_assets` | — | Base assets expanded via whitelist templates (e.g. `["BTC", "ETH"]`) | +| `max_storage_mb` | `4096` | Hard cap on database file size (MB); `0` = unlimited. | +| `data_retention_hours` | `168` | Trades older than this are purged; `0` = keep all indefinitely. | + +### Advanced + +| Option | Default | Description | +| --------------------- | ---------------------- | --------------------------------------------------------------------------- | +| `discovery_mode` | `true` | Fetch metadata for all exchange variants so `/exchanges` is fully populated | +| `tls_domain` | `"flowsurface-server"` | Domain in the self-signed TLS cert's SAN (only needed for verified TLS) | +| `flush_interval_ms` | `2000` | How often buffered trades are written to disk (ms); higher = fewer writes | +| `max_buffered_trades` | `200000` | Max trades in memory buffer before dropping (OOM guard) | ### Whitelist templates @@ -89,9 +96,8 @@ When binding to a **non-loopback** address, the server: cat data/.auth_token ``` -The token is reused across restarts. To set a specific token manually, -add `auth_token = "your-token"` to `config.toml`, or set the -`AUTH_TOKEN` environment variable (via `.env` or the environment). +The token is reused across restarts. To set a specific token set the `AUTH_TOKEN` environment variable +(via `.env` or the environment). ## Remote deployment (VPS) @@ -141,13 +147,109 @@ For local-only use, keep `bind_address = "127.0.0.1:8080"`: All other endpoints require `Authorization: Bearer ` when auth is configured. -| Method | Path | Auth | Description | -| ------ | --------------- | -------- | ---------------------------------------------------------- | -| GET | `/status` | ✗ public | Server uptime, DB connectivity check | -| GET | `/exchanges` | required | Available ticker symbols per exchange | -| GET | `/pairs` | required | Configured pairs with time bounds & tracked count | -| GET | `/trades` | required | Trade data (filtered by venue, symbol, time range) | -| GET | `/trades.arrow` | required | Trade data as [Arrow IPC](https://arrow.apache.org) stream | +| Method | Path | Auth | Description | +| ------ | --------------- | -------- | -------------------------------------------------- | +| GET | `/status` | ✗ public | Server uptime, DB connectivity check | +| GET | `/exchanges` | required | Available ticker symbols per exchange | +| GET | `/pairs` | required | Configured pairs with time bounds & tracked count | +| GET | `/trades` | required | Trade data (filtered by venue, symbol, time range) | +| GET | `/trades.arrow` | required | Trade data as Arrow IPC stream | + +### GET /status + +Returns the server health status. No authentication required — suitable for +load balancer health checks. + +#### Response fields + +| Field | Type | Description | +| ------------- | ------ | ----------------------------- | +| `status` | string | Always `"ok"` while running | +| `uptime_secs` | int | Seconds since server start | +| `db_ok` | bool | `true` if DuckDB is reachable | + +#### Example + +```bash +curl http://127.0.0.1:8080/status +``` + +```json +{ + "status": "ok", + "uptime_secs": 7, + "db_ok": true +} +``` + +### GET /exchanges + +Returns every ticker symbol discovered on each exchange, grouped by +canonical exchange name. Useful for browsing available tickers before +configuring the whitelist. + +```bash +curl -H "Authorization: Bearer " \ + http://127.0.0.1:8080/exchanges +``` + +```json +{ + "exchanges": { + "Binance Linear": ["BTCUSDT", "ETHUSDT", ...], + "Binance Spot": ["BTCUSDT", "ETHUSDT", ...], + "Bybit Linear": ["BTCUSDT", ...], + ... + } +} + +``` + +### GET /pairs + +Returns all configured pairs with their stored time ranges. Pairs that have +been configured but have not yet received any trades appear with `earliest` +and `latest` as `null`. + +#### Response fields + +| Field | Type | Description | +| --------------- | ----- | -------------------------------- | +| `pairs` | array | Array of tracked pair objects | +| `tracked_count` | int | Total number of configured pairs | + +Each pair object: + +| Field | Type | Description | +| ---------- | ------ | ----------------------------------------- | +| `ticker` | string | Ticker ID (`Exchange:pair`) | +| `earliest` | int | Unix ms of oldest stored trade, or `null` | +| `latest` | int | Unix ms of newest stored trade, or `null` | + +#### Example + +```bash +curl -H "Authorization: Bearer " \ + http://127.0.0.1:8080/pairs +``` + +```json +{ + "pairs": [ + { + "ticker": "BinanceSpot:btcusdt", + "earliest": 1783873393043, + "latest": 1784372262447 + }, + { + "ticker": "HyperliquidLinear:btcusdc", + "earliest": 1784372175005, + "latest": 1784372262010 + } + ], + "tracked_count": 18 +} +``` ### GET /trades @@ -207,8 +309,7 @@ format** payload (`Content-Type: application/vnd.apache.arrow.stream`) with 4 columns: `ts (int64)`, `price (float64)`, `qty (float64)`, `is_sell (bool)`. -This is ideal for high-volume data transfer into data-science tools -(Polars, Pandas, Julia, etc.) that support Arrow natively. +This is ideal for high-volume data transfer to clients that support Arrow natively. | Param | Type | Description | | -------- | ------ | -------------------------------------------- | @@ -218,26 +319,3 @@ This is ideal for high-volume data transfer into data-science tools | `from` | int | Unix ms lower bound (inclusive) | | `to` | int | Unix ms upper bound (inclusive) | | `limit` | int | Max records (default 100 000, max 1 000 000) | - -### GET /exchanges - -Returns every ticker symbol discovered on each exchange, grouped by -canonical exchange name. Useful for browsing available tickers before -configuring the whitelist. - -```bash -curl -H "Authorization: Bearer " \ - http://127.0.0.1:8080/exchanges -``` - -```json -{ - "exchanges": { - "Binance Linear": ["BTCUSDT", "ETHUSDT", ...], - "Binance Spot": ["BTCUSDT", "ETHUSDT", ...], - "Bybit Linear": ["BTCUSDT", ...], - ... - } -} - -``` diff --git a/config.example.toml b/config.example.toml index b8b35ee..1cafa02 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1,43 +1,32 @@ -bind_address = "127.0.0.1:8080" -data_dir = "./data" +# Directory for DuckDB database, auth token, and TLS certs. +# Relative paths are resolved from the config file's directory. +# data_dir = "data" -# Domain name for the self-signed TLS certificate's SAN (Subject Alternative Names). -# Ignored on loopback binds (plain HTTP). Default: "flowsurface-server". -# tls_domain = "data.example.com" +# ── Network ───────────────────────────────────────────────────── +# Where the API listens. Change to "0.0.0.0:8080" for remote +# access (auto-generates TLS cert + auth token). +bind_address = "127.0.0.1:8080" -# Trade batch-flush interval in milliseconds. -# Lower values reduce data loss on crash/failure; higher values are more I/O-efficient -# (fewer fsyncs, larger columnar batches). Default: 2000 (2 seconds). -# flush_interval_ms = 2000 +# ── Storage ────────────────────────────────────────────────────── +# Hard cap on total DuckDB database file size in megabytes. +# When exceeded the oldest trades are purged during cleanup — even +# if they're within the time-based retention window. +# Set to 0 to disable the size cap. +# Default: 4096 (4 GiB). +max_storage_mb = 4096 -# Data retention period in hours. Trades older than this will be purged -# on startup (and periodically while running). Default: 48 (2 days). -# data_retention_hours = 48 +# Data retention period in hours. Trades older than this are +# purged on startup and periodically while running. +# Set to 0 to keep all trades indefinitely (disk-permitting). +# Default: 168 (7 days). +data_retention_hours = 168 -# Base assets to track — expanded via the whitelist templates below. -# The server fetches exchange metadata and resolves pairs at startup. +# ── Pairs to track ────────────────────────────────────────────── +# Expand base_assets × quote_assets from the whitelist templates +# below. The server resolves exchange-specific ticker strings +# (handling separators, _PERP, -SWAP suffixes, etc.). base_assets = ["BTC", "ETH"] -# Whitelist templates: per-venue, per-market-kind list of quote assets. -# To exclude a venue entirely, omit it from the whitelist. -# To exclude a market kind for a venue, omit that key. -# -# Tip: `http(s):///exchanges` shows all available tickers that can be tracked. -# use `[""]` (empty string) as a wildcard to track all `base_assets` regardless of the quote on that venue+market -# -# Discovery mode: fetches and caches metadata for ALL supported exchanges on startup -# so `/exchanges` is fully populated. Lets you browse available tickers -# before deciding what to track. Default: true. -# discovery_mode = true - -# Maximum trades to buffer in memory before dropping to prevent OOM. -# Tune based on your host RAM and risk tolerance: -# 200k (~20-40 MB) — safe for 1 GB hosts -# 1.2M (~120-240 MB) — ~2 min buffer at 10k trades/sec -# Higher values reduce data-loss during DB outages but use more memory. -# Default: 200000. -# max_buffered_trades = 200000 - [whitelist.binance] spot = ["USDT"] linear = ["USDT", "USDC"] @@ -52,4 +41,28 @@ linear = ["USDC"] [whitelist.okex] spot = ["USDT"] -linear = ["USDT"] \ No newline at end of file +linear = ["USDT"] + +# ── Advanced options ───────────────────────────────────────────── +# Most users can leave these at their defaults. + +# Discovery mode: fetches metadata for ALL supported exchanges on +# startup so /exchanges is fully populated. +# If false, only fetch metadata for whitelisted venues +# Default: true. +# discovery_mode = true + +# Domain name for the self-signed TLS certificate's SAN. +# Only needed when connecting via a domain with verified TLS. +# tls_domain = "data.example.com" + +# Trade batch-flush interval in milliseconds. +# Higher values = fewer disk writes (more I/O efficient). +# Lower values = trades appear in the DB sooner (at the cost of more writes). +# Default: 2000 (2 seconds). +# flush_interval_ms = 2000 + +# Maximum trades to buffer in memory before dropping (OOM guard). +# 200k ≈ 20-40 MB — safe for 1 GB hosts. +# Default: 200000. +# max_buffered_trades = 200000 \ No newline at end of file diff --git a/dist-workspace.toml b/dist-workspace.toml new file mode 100644 index 0000000..95cdfe9 --- /dev/null +++ b/dist-workspace.toml @@ -0,0 +1,15 @@ +[workspace] +members = ["cargo:."] + +# Config for 'dist' +[dist] +# The preferred dist version to use in CI (Cargo.toml SemVer syntax) +cargo-dist-version = "0.32.0" +# CI backends to support +ci = "github" +# The installers to generate for each app +installers = [] +# Target platforms to build apps for (Rust target-triple syntax) +targets = ["aarch64-apple-darwin", "aarch64-unknown-linux-gnu", "x86_64-apple-darwin", "x86_64-unknown-linux-gnu", "x86_64-pc-windows-msvc"] +# Extra static files to include in the archive (README, LICENSE are auto-included) +include = ["config.example.toml"] diff --git a/src/api.rs b/src/api.rs index e363ce4..c63e565 100644 --- a/src/api.rs +++ b/src/api.rs @@ -17,7 +17,10 @@ use flowsurface_exchange::{ }; use serde::{Deserialize, Serialize}; -use crate::storage::{PairInfo, Storage}; +use crate::{ + config::BearerToken, + storage::{PairInfo, Storage}, +}; #[derive(Serialize)] #[serde(untagged)] @@ -100,7 +103,7 @@ pub fn exchange_from_venue_market(venue: &str, market: &str) -> Option { pub struct Server { pub storage: Storage, pub startup: Instant, - pub auth_token: Option, + pub auth_token: Option, /// The tickers configured at startup. /// Used by `/pairs` to include pairs that have not yet received trades. pub configured_pairs: Vec, @@ -115,7 +118,7 @@ pub struct Server { impl Server { pub fn new( storage: Storage, - auth_token: Option, + auth_token: Option, configured_pairs: Vec, available_tickers: HashMap>, tls_config: Option, @@ -146,9 +149,7 @@ impl Server { .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let expected = format!("Bearer {expected_token}"); - - if !provided.eq_ignore_ascii_case(&expected) { + if !expected_token.is_valid_authorization(provided) { let truncated: String = provided.chars().take(20).collect(); tracing::warn!( "Auth failure from {}: expected valid Bearer token, got '{truncated}'", @@ -312,12 +313,7 @@ impl Server { /// /// Uses plain HTTP for loopback addresses, HTTPS with a self-signed /// certificate for non-loopback (remote) binds. Exits on bind failure. - pub async fn serve(self: Arc, bind_address: &str) -> tokio::task::JoinHandle<()> { - let addr: SocketAddr = bind_address.parse().unwrap_or_else(|e| { - tracing::error!("Invalid bind_address '{bind_address}': {e}"); - std::process::exit(1); - }); - + pub async fn serve(self: Arc, bind_address: SocketAddr) -> tokio::task::JoinHandle<()> { let tls_config = self.tls_config.clone(); // Public routes — no auth required @@ -341,17 +337,17 @@ impl Server { tokio::spawn(async move { if let Some(cfg) = tls_config { - tracing::info!("Starting HTTPS API on {addr}"); - axum_server::bind_rustls(addr, cfg) + tracing::info!("Starting HTTPS API on {bind_address}"); + axum_server::bind_rustls(bind_address, cfg) .serve(router.into_make_service_with_connect_info::()) .await .unwrap(); } else { - tracing::info!("Starting HTTP API on {addr}"); - let listener = tokio::net::TcpListener::bind(addr) + tracing::info!("Starting HTTP API on {bind_address}"); + let listener = tokio::net::TcpListener::bind(bind_address) .await .unwrap_or_else(|e| { - tracing::error!("Failed to bind to {addr}: {e}"); + tracing::error!("Failed to bind to {bind_address}: {e}"); std::process::exit(1); }); axum::serve( diff --git a/src/cleanup.rs b/src/cleanup.rs new file mode 100644 index 0000000..1d7fe00 --- /dev/null +++ b/src/cleanup.rs @@ -0,0 +1,341 @@ +use std::time::Duration; + +use anyhow::Result; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +use crate::{ + config::{RetentionHours, StorageBytes}, + storage::Storage, +}; + +/// Floor on delay between cleanup passes; prevents tight retry loops +/// when `last_cleanup` is stale (first run or previous pass failed). +const MIN_CLEANUP_DELAY: Duration = Duration::from_secs(60); + +/// Wake interval for the size-cap check. Only matters when a size cap +/// is configured and the retention window is longer than this. +const SIZE_CHECK_INTERVAL: Duration = Duration::from_secs(300); + +/// Safety ceiling for the size-based purge loop (200 × 100k = 20M rows). +const MAX_PURGE_ITERATIONS: usize = 200; + +/// Conservative estimate of on-disk bytes per trade row. Used only as +/// a fast-path threshold; the authoritative check is the real file +/// size after `CHECKPOINT`. +const EST_BYTES_PER_ROW: u64 = 50; + +/// On small-to-medium caps use a 20 % headroom so the purge leaves +/// breathing room. On very large caps the percentage would waste +/// too much space, so we cap the absolute headroom at 10 GiB +const MAX_HEADROOM: StorageBytes = StorageBytes::from_bytes(10 * 1024 * 1024 * 1024); + +/// Smallest sane target after headroom subtraction — prevents +/// pathological behaviour on tiny caps. +const MIN_TARGET: StorageBytes = StorageBytes::from_bytes(100 * 1024 * 1024); + +/// Compute the headroom to leave free below the cap. +/// Returns `cap / 5` (20 %) but at most `MAX_HEADROOM`. +fn purge_headroom(cap: StorageBytes) -> StorageBytes { + let pct = cap.as_bytes() / 5; + if pct > MAX_HEADROOM.as_bytes() { + MAX_HEADROOM + } else { + StorageBytes::from_bytes(pct) + } +} + +/// Batch size for each purge iteration — ~10% of the cap, clamped to +/// [1 000, 100 000]. +fn purge_batch_size(max_bytes: u64) -> i64 { + const MIN_ROWS: i64 = 1_000; + const MAX_ROWS: i64 = 100_000; + let rows = (max_bytes / 10 / EST_BYTES_PER_ROW) as i64; + rows.clamp(MIN_ROWS, MAX_ROWS) +} + +#[derive(Debug, Clone, Copy)] +pub struct CleanupConfig { + /// Trades older than this are purged. + pub retention_hours: RetentionHours, + /// Optional hard cap on total DB+WAL size in bytes. + pub max_storage_bytes: Option, +} + +impl CleanupConfig { + /// Derive from user-facing settings. `max_storage_mb` of `0` or + /// `None` disables the cap. Returns an error if + /// `retention_hours` is zero. + pub fn from_config(retention_hours: u64, max_storage_mb: Option) -> anyhow::Result { + let retention_hours = RetentionHours::new(retention_hours)?; + let max_storage_bytes = max_storage_mb + .filter(|&mb| mb > 0) + .map(StorageBytes::from_mb); + Ok(Self { + retention_hours, + max_storage_bytes, + }) + } +} + +#[derive(Clone)] +pub struct CleanupScheduler { + storage: Storage, + config: CleanupConfig, +} + +impl CleanupScheduler { + /// Create the scheduler and check the startup guard: if the database + /// is > 2× the configured cap the process exits (this is a + /// misconfiguration that would take too long to recover from at + /// startup). + /// + /// Returns an error when the storage size cannot be read at all. + pub fn new(storage: &Storage, config: CleanupConfig) -> anyhow::Result { + if let Some(max) = config.max_storage_bytes { + tracing::info!( + "Storage hard cap enabled: {} MB (will purge oldest trades when exceeded)", + max.as_mb(), + ); + + match storage.current_storage_bytes() { + Ok(current) if current > max.saturating_mul(2) => { + anyhow::bail!( + "Database is {} MB — more than 2x the configured \ + max_storage_mb cap ({} MB). Purging that much data \ + at startup would take too long; this likely indicates \ + a misconfiguration. Either raise max_storage_mb \ + (e.g. to {} MB or higher) or manually shrink the \ + database and restart.", + current.as_mb(), + max.as_mb(), + current.as_mb().saturating_add(1), + ); + } + Ok(current) if current > max => { + tracing::warn!( + "Database is {} MB — above the {} MB cap; \ + startup cleanup will purge oldest trades.", + current.as_mb(), + max.as_mb(), + ); + } + Err(e) => { + anyhow::bail!("Failed to check storage size at startup: {e:#}"); + } + _ => {} + } + } + + Ok(Self { + storage: storage.clone(), + config, + }) + } + + /// Run one cleanup pass: time-based retention, optional size-cap + /// purge, checkpoint, and record `last_cleanup`. + /// + /// Returns `Some(last_cleanup_ts)` on success, or `None` if the + /// pass failed (e.g. time-based purge errored or recording the + /// timestamp failed). + pub fn run_pass(&self) -> Option { + let mut any_deleted = false; + let mut time_cleanup_ok = false; + + match self.storage.purge_old_trades(self.config.retention_hours) { + Ok(n) => { + time_cleanup_ok = true; + if n > 0 { + tracing::info!( + "Cleaned up {n} trade(s) older than {}h", + self.config.retention_hours.as_hours() + ); + any_deleted = true; + } + } + Err(e) => { + tracing::error!("Data retention cleanup failed: {e:#}"); + } + } + + if let Some(max_bytes) = self.config.max_storage_bytes { + match self.purge_oldest_trades_until_below(max_bytes) { + Ok(n) if n > 0 => any_deleted = true, + Err(e) => tracing::error!("Storage-cap cleanup failed: {e:#}"), + _ => {} + } + } + + if any_deleted && let Err(e) = self.storage.run_checkpoint() { + tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); + } + + if time_cleanup_ok { + match self.storage.record_cleanup() { + Ok(ts) => return Some(ts), + Err(e) => tracing::warn!("Failed to record last_cleanup timestamp: {e:#}"), + } + } + None + } + + /// Delete oldest trades until on-disk size is estimated to be under the cap. + /// + /// To avoid per-iteration CHECKPOINT overhead we use a two-phase approach: + /// 1. **Row-count estimate loop** — delete batches until the estimated row + /// count is below the headroom-adjusted threshold. No CHECKPOINTs here. + /// 2. **CHECKPOINT once**, then verify the real file size. If still over + /// the absolute cap we log a warning — the next periodic pass will retry. + /// + /// Targets `cap - headroom` so there is breathing room for incoming + /// trades between 5-minute checks without wasting space on large caps. + fn purge_oldest_trades_until_below(&self, max_bytes: StorageBytes) -> Result { + let headroom = purge_headroom(max_bytes); + let target_bytes = max_bytes + .as_bytes() + .saturating_sub(headroom.as_bytes()) + .max(MIN_TARGET.as_bytes()); + let target_bytes = StorageBytes::from_bytes(target_bytes); + let max_est_rows = target_bytes.as_bytes().saturating_div(EST_BYTES_PER_ROW); + let batch_size = purge_batch_size(target_bytes.as_bytes()); + let mut total_deleted = 0u64; + let mut converged = false; + + for _ in 0..MAX_PURGE_ITERATIONS { + // Cheap row-count check (no CHECKPOINT needed). + if self.storage.count_trades()? <= max_est_rows { + converged = true; + break; + } + let deleted = self.storage.delete_oldest_trades_batch(batch_size)?; + if deleted == 0 { + converged = true; + break; + } + total_deleted += deleted; + } + + if total_deleted > 0 { + self.storage.run_checkpoint()?; + let remaining_rows = self.storage.count_trades()?; + + if converged { + let current = self.storage.current_storage_bytes()?; + if current <= max_bytes { + tracing::info!( + "Cleaned up {total_deleted} trade(s) to keep storage under \ + {} MB cap (now ~{} MB)", + max_bytes.as_mb(), + current.as_mb(), + ); + } else { + tracing::warn!( + "Storage ({} MB) still exceeds {} MB cap after cleanup. \ + The next periodic pass will retry.", + current.as_mb(), + max_bytes.as_mb(), + ); + } + } else { + tracing::warn!( + "Size-cap purge did not converge after \ + {MAX_PURGE_ITERATIONS} iterations (deleted {total_deleted} \ + rows); storage may still exceed the {} MB cap", + max_bytes.as_mb(), + ); + } + + if should_vacuum(total_deleted, remaining_rows) { + if let Err(e) = self.storage.vacuum() { + tracing::warn!( + "VACUUM after size-cap purge failed (data is \ + correct, but filesystem space wasn't reclaimed): \ + {e:#}" + ); + } else { + tracing::debug!( + "VACUUMed database after purging {total_deleted} \ + rows ({remaining_rows} remaining)" + ); + } + } + } + + Ok(total_deleted) + } + + /// Spawn the periodic background cleanup task. + /// + /// A one-shot startup pass must have been run beforehand (via + /// [`run_startup_pass`](Self::run_startup_pass)) so that + /// `last_cleanup` is initialised. The loop then computes the + /// next wake-up from `last_cleanup` and only polls when work + /// is actually due (or every `SIZE_CHECK_INTERVAL` when a size + /// cap is configured). + pub fn spawn(self, last_cleanup: Option, shutdown: CancellationToken) -> JoinHandle<()> { + let storage = self.storage.clone(); + let config = self.config; + + let retention_ms = config.retention_hours.as_millis(); + + tokio::spawn(async move { + let mut last_cleanup = last_cleanup; + + loop { + let time_delay = Self::delay_until_next_cleanup(last_cleanup, retention_ms); + let delay = + if config.max_storage_bytes.is_some() && SIZE_CHECK_INTERVAL < time_delay { + SIZE_CHECK_INTERVAL + } else { + time_delay + } + .max(MIN_CLEANUP_DELAY); + + tokio::select! { + biased; + _ = shutdown.cancelled() => { + tracing::info!("Periodic cleanup shut down."); + break; + } + _ = tokio::time::sleep(delay) => { + let result = tokio::task::spawn_blocking({ + let storage = storage.clone(); + move || { + CleanupScheduler { storage, config }.run_pass() + } + }) + .await + .unwrap_or(None); + if let Some(ts) = result { + last_cleanup = Some(ts); + } + } + } + } + }) + } + + fn delay_until_next_cleanup(last_cleanup_ms: Option, retention_ms: i64) -> Duration { + let now_ms = Storage::now_ms(); + + let Some(anchor_ms) = last_cleanup_ms else { + // No recorded timestamp — retry quickly (caller applies + // `MIN_CLEANUP_DELAY`). + return Duration::ZERO; + }; + + let next_ms = anchor_ms + retention_ms; + if next_ms <= now_ms { + Duration::ZERO + } else { + Duration::from_millis((next_ms - now_ms) as u64) + } + } +} + +/// Whether VACUUM is worth running after a purge — only when we've +/// freed at least ~25% of the remaining data. +const fn should_vacuum(total_deleted: u64, remaining_rows: u64) -> bool { + remaining_rows == 0 || total_deleted >= remaining_rows / 4 +} diff --git a/src/config.rs b/src/config.rs index 2eabcc5..29c40b5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,8 +1,13 @@ +use std::collections::HashMap; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; + use anyhow::{Context, Result}; use clap::Parser; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::path::{Path, PathBuf}; + +use std::fmt; +use std::str::FromStr; /// Whitelist templates: venue → market_kind → list of quote assets. /// @@ -23,7 +28,7 @@ pub struct Args { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Config { /// Socket address to bind the HTTP API (e.g. `127.0.0.1:8080`). - pub bind_address: String, + pub bind_address: SocketAddr, /// Directory where the DuckDB database file will be stored. pub data_dir: String, /// Optional bearer-token required on all API requests. @@ -31,8 +36,12 @@ pub struct Config { /// /// The server uses HTTPS with a self-signed certificate (generated /// on first boot), so the token is always encrypted in transit. - #[serde(default, skip_serializing)] - pub auth_token: Option, + /// + /// When generated automatically the token is stored in + /// `data_dir / .auth_token` so that restarts reuse the same token. + /// Not settable via `config.toml` — use the `AUTH_TOKEN` env var instead. + #[serde(skip)] + pub auth_token: Option, // ── Pair tracking ─────────────────────────────────────────── /// Base assets to expand via the whitelist templates (e.g. `["btc", "eth"]`). @@ -78,6 +87,20 @@ pub struct Config { /// Default: 200_000 (~20–40 MB depending on symbol length). #[serde(default = "default_max_buffered_trades")] pub max_buffered_trades: usize, + + /// Optional hard cap on total DuckDB storage (main DB + WAL) in + /// megabytes. When the combined file size exceeds this value the + /// oldest trades are purged during cleanup — even if they're within + /// the time-based retention window. + /// + /// Use this to prevent the database from filling the disk on + /// constrained hosts. Default: `None` (no size cap). + /// + /// Tip: set this to ~50-80 % of your available disk space so the + /// server leaves room for system files, logs, and burst. + /// Default: `4096` (4 GiB). + #[serde(default = "default_max_storage_mb")] + pub max_storage_mb: Option, } const fn default_flush_interval() -> u64 { @@ -85,7 +108,7 @@ const fn default_flush_interval() -> u64 { } const fn default_data_retention_hours() -> u64 { - 48 + 168 } const fn default_true() -> bool { @@ -100,6 +123,10 @@ const fn default_max_buffered_trades() -> usize { 200_000 } +const fn default_max_storage_mb() -> Option { + Some(4096) +} + impl Config { /// Return the default config template as a commented TOML string. pub fn template() -> &'static str { @@ -114,7 +141,7 @@ impl Config { toml::from_str(&content).with_context(|| format!("parsing {}", path.display()))?; if let Ok(token) = std::env::var("AUTH_TOKEN") - && !token.is_empty() + && let Some(token) = BearerToken::new(token) { cfg.auth_token = Some(token); } @@ -128,53 +155,51 @@ impl Config { } /// Resolve the configuration file path. + /// + /// Defaults to next to the binary so the entire app is portable in + /// a single directory. pub fn resolve_path(override_path: Option) -> PathBuf { if let Some(path) = override_path { return path; } - if let Ok(exe) = std::env::current_exe() - && let Some(parent) = exe.parent() - { - let candidate = parent.join("config.toml"); - if candidate.exists() { - return candidate; + if let Ok(exe) = std::env::current_exe() { + if let Some(parent) = exe.parent() { + return parent.join("config.toml"); } } PathBuf::from("config.toml") } pub fn resolve_auth_token(&mut self) -> anyhow::Result<()> { - let addr: std::net::SocketAddr = self - .bind_address - .parse() - .with_context(|| format!("invalid bind_address '{}'", self.bind_address))?; - - if addr.ip().is_loopback() { + if self.bind_address.ip().is_loopback() { return Ok(()); } let token_file = std::path::PathBuf::from(&self.data_dir).join(".auth_token"); - let on_disk = std::fs::read_to_string(&token_file) + let on_disk_raw = std::fs::read_to_string(&token_file) .ok() - .map(|s| s.trim().to_string()) + .map(|s| s.trim().to_owned()) .filter(|s| !s.is_empty()); + let on_disk_token = on_disk_raw + .as_deref() + .and_then(|s| BearerToken::new(s.to_owned())); - let token = match (self.auth_token.clone(), on_disk.clone()) { + let token = match (self.auth_token.clone(), on_disk_token) { (Some(explicit), _) => explicit, // env/config wins (None, Some(existing)) => existing, // reuse what's on disk (None, None) => generate_token(), // nothing anywhere — mint one }; - if on_disk.as_deref() != Some(token.as_str()) { + if on_disk_raw.as_deref() != Some(token.as_str()) { std::fs::create_dir_all(&self.data_dir) .with_context(|| format!("creating data dir '{}'", self.data_dir))?; - std::fs::write(&token_file, &token) + std::fs::write(&token_file, token.as_str()) .with_context(|| format!("writing {}", token_file.display()))?; crate::tls::restrict_permissions(&token_file); tracing::info!( "Auth token → {}\n Token starts with: {}… (run `cat {}` to view full token)", token_file.display(), - &token[..4.min(token.len())], + &token.as_str()[..4.min(token.as_str().len())], token_file.display(), ); } @@ -222,9 +247,145 @@ impl Config { } } +/// A Bearer token used to authenticate API requests. +/// +/// Constructed via [`FromStr`] (or [`BearerToken::new`]) which rejects +/// empty strings. The [`Display`] implementation outputs `[REDACTED]` +/// to prevent accidental leakage in logs. +/// +/// # Example +/// +/// ```ignore +/// let token: BearerToken = "my-secret-token".parse()?; +/// assert!(token.is_valid_authorization("Bearer my-secret-token")); +/// assert!(!token.is_valid_authorization("Bearer wrong-token")); +/// ``` +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(try_from = "String")] +pub struct BearerToken(String); + +impl BearerToken { + /// Create a new `BearerToken`, returning `None` if `raw` is empty. + pub fn new(raw: String) -> Option { + if raw.is_empty() { + None + } else { + Some(Self(raw)) + } + } + + /// Check whether `authorization_header` matches `"Bearer {token}"` + /// (case-insensitive). + pub fn is_valid_authorization(&self, authorization_header: &str) -> bool { + let expected = format!("Bearer {}", self.0); + authorization_header.eq_ignore_ascii_case(&expected) + } + + /// Return the raw token string (for writing to disk, etc.). + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl FromStr for BearerToken { + type Err = &'static str; + + fn from_str(s: &str) -> Result { + if s.is_empty() { + Err("Bearer token must not be empty") + } else { + Ok(Self(s.to_owned())) + } + } +} + +impl TryFrom for BearerToken { + type Error = &'static str; + + fn try_from(value: String) -> Result { + value.parse() + } +} + +impl fmt::Display for BearerToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "[REDACTED]") + } +} + +impl AsRef for BearerToken { + fn as_ref(&self) -> &str { + &self.0 + } +} + +impl PartialEq for BearerToken { + fn eq(&self, other: &Self) -> bool { + // Constant-time comparison would be better, but for a self-hosted + // internal API the timing leak is negligible. + self.0 == other.0 + } +} + +/// Retention period expressed in hours. A value of `0` means +/// **unlimited** — no time-based purges. +#[derive(Debug, Clone, Copy)] +pub struct RetentionHours(u64); + +impl RetentionHours { + /// Create a `RetentionHours`. `0` is accepted and means unlimited + /// (time-based purges are skipped). + pub fn new(hours: u64) -> anyhow::Result { + Ok(Self(hours)) + } + + /// The raw hour count. + pub fn as_hours(self) -> u64 { + self.0 + } + + /// The equivalent span in milliseconds (as a signed value for SQL). + pub fn as_millis(self) -> i64 { + (self.0 as i64) * 3_600_000 + } +} + +/// A storage size expressed in bytes. Provides convenience conversions +/// to megabytes to avoid sprinkling `(1024 * 1024)` throughout the code. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct StorageBytes(u64); + +impl StorageBytes { + /// Construct from a byte count. + pub const fn from_bytes(bytes: u64) -> Self { + Self(bytes) + } + + /// Construct from a megabyte count (clamped to `u64::MAX` on overflow). + pub fn from_mb(mb: u64) -> Self { + Self(mb.saturating_mul(1024 * 1024)) + } + + /// The raw byte count. + pub const fn as_bytes(self) -> u64 { + self.0 + } + + /// The size in whole megabytes (truncated). + pub fn as_mb(self) -> u64 { + self.0 / (1024 * 1024) + } + + /// Saturating multiplication (returns `StorageBytes`). + pub fn saturating_mul(self, rhs: u64) -> Self { + Self(self.0.saturating_mul(rhs)) + } +} + /// Generate a random 256-bit hex token. -fn generate_token() -> String { +fn generate_token() -> BearerToken { let mut buf = [0u8; 32]; getrandom::getrandom(&mut buf).expect("failed to get random bytes"); - buf.iter().map(|b| format!("{b:02x}")).collect() + let hex: String = buf.iter().map(|b| format!("{b:02x}")).collect(); + BearerToken::new(hex).expect("generated hex token is never empty") } diff --git a/src/main.rs b/src/main.rs index 8c51b1d..8e9b6d3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,13 @@ mod api; +mod cleanup; mod config; mod discovery; mod ingestion; mod storage; mod tls; -use std::path::PathBuf; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; use std::sync::Arc; use clap::Parser; @@ -18,7 +20,7 @@ use flowsurface_exchange::adapter::{AdapterHandles, Venue}; use flowsurface_exchange::{Ticker, TickerInfo}; use crate::api::Server; -use crate::config::{Args, Config}; +use crate::config::{Args, BearerToken, Config}; use crate::storage::Storage; #[tokio::main] @@ -45,7 +47,16 @@ async fn main() { std::process::exit(1); } - let app = App::new(&config).await; + let data_dir = if Path::new(&config.data_dir).is_relative() { + config_path + .parent() + .expect("config path has no parent") + .join(&config.data_dir) + } else { + PathBuf::from(&config.data_dir) + }; + + let app = App::new(&config, &data_dir).await; let handles = app.serve().await; handles.shutdown().await; } @@ -55,24 +66,34 @@ struct App { adapter_handles: AdapterHandles, resolved_pairs: Vec, metadata_cache: discovery::MetadataCache, - bind_address: String, - auth_token: Option, + bind_address: SocketAddr, + auth_token: Option, flush_interval: std::time::Duration, max_buffered_trades: usize, - data_retention_hours: u64, + cleanup_scheduler: cleanup::CleanupScheduler, tls_config: Option, } impl App { /// Open storage, resolve configured pairs, persist ticker metadata. - async fn new(config: &Config) -> Self { - let data_dir = PathBuf::from(&config.data_dir); - let storage = Storage::open(&data_dir).unwrap_or_else(|e| { + async fn new(config: &Config, data_dir: &Path) -> Self { + let storage = Storage::open(data_dir).unwrap_or_else(|e| { tracing::error!("Failed to initialise storage: {e:#}"); std::process::exit(1); }); - storage.run_cleanup(config.data_retention_hours); + let cleanup_config = + cleanup::CleanupConfig::from_config(config.data_retention_hours, config.max_storage_mb) + .unwrap_or_else(|e| { + tracing::error!("Invalid cleanup configuration: {e:#}"); + std::process::exit(1); + }); + + let cleanup_scheduler = cleanup::CleanupScheduler::new(&storage, cleanup_config) + .unwrap_or_else(|e| { + tracing::error!("{e:#}"); + std::process::exit(1); + }); let whitelist = config.resolve_whitelist(); if !config.discovery_mode && (whitelist.is_empty() || config.base_assets.is_empty()) { @@ -128,16 +149,12 @@ impl App { } // Only generate TLS cert for non-loopback addresses. - let addr: std::net::SocketAddr = config - .bind_address - .parse() - .expect("bind_address already validated"); - - let tls_config = if addr.ip().is_loopback() { + let tls_config = if config.bind_address.ip().is_loopback() { None } else { let tls_domain = config.tls_domain.clone(); - let bind_ip = (!addr.ip().is_unspecified()).then_some(addr.ip()); + let bind_ip = + (!config.bind_address.ip().is_unspecified()).then_some(config.bind_address.ip()); let cert_path = data_dir.join("cert.pem"); let key_path = data_dir.join("key.pem"); @@ -194,12 +211,12 @@ impl App { adapter_handles, resolved_pairs, metadata_cache, - bind_address: config.bind_address.clone(), + bind_address: config.bind_address, auth_token: config.auth_token.clone(), flush_interval: std::time::Duration::from_millis(config.flush_interval_ms), max_buffered_trades: config.max_buffered_trades, - data_retention_hours: config.data_retention_hours, tls_config, + cleanup_scheduler, } } @@ -208,6 +225,13 @@ impl App { let (trade_tx, trade_rx) = mpsc::unbounded_channel::(); let shutdown = CancellationToken::new(); + let cleanup_last_run = tokio::task::spawn_blocking({ + let scheduler = self.cleanup_scheduler.clone(); + move || scheduler.run_pass() + }) + .await + .unwrap_or(None); + let flusher = self.storage.spawn_batch_flusher( trade_rx, self.flush_interval, @@ -215,8 +239,8 @@ impl App { ); let _cleanup = self - .storage - .spawn_periodic_cleanup(self.data_retention_hours, shutdown.child_token()); + .cleanup_scheduler + .spawn(cleanup_last_run, shutdown.child_token()); let ingest = ingestion::start_all_ingest_tasks( &self.resolved_pairs, @@ -237,7 +261,7 @@ impl App { available_tickers, self.tls_config, )); - let server_handle = server.serve(&self.bind_address).await; + let server_handle = server.serve(self.bind_address).await; AppHandles { shutdown, diff --git a/src/storage.rs b/src/storage.rs index 2a8f133..924e0fc 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -1,6 +1,6 @@ use anyhow::{Context, Result}; use duckdb::{Appender, Connection}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; @@ -9,9 +9,9 @@ use flowsurface_exchange::unit::{price::Price, qty::Qty}; use flowsurface_exchange::{Ticker, TickerInfo, UnixMs}; use crate::api::{AnnotatedTrade, TradeQuery}; +use crate::config::{RetentionHours, StorageBytes}; use tokio::sync::mpsc; use tokio::task::JoinHandle; -use tokio_util::sync::CancellationToken; /// Information about a tracked pair with the timestamp range stored. #[derive(Debug, Clone, Copy, serde::Serialize)] @@ -33,6 +33,7 @@ pub struct PairInfo { #[derive(Clone)] pub struct Storage { db: Arc>, + data_dir: PathBuf, } impl Storage { @@ -87,9 +88,109 @@ impl Storage { Ok(Self { db: Arc::new(parking_lot::Mutex::new(root)), + data_dir: data_dir.to_path_buf(), }) } + /// Return the size (in bytes) of the main database file plus the + /// WAL file. If a file does not (yet) exist its size is counted as 0. + pub fn current_storage_bytes(&self) -> Result { + let db_path = self.data_dir.join("trades.duckdb"); + let wal_path = self.data_dir.join("trades.duckdb.wal"); + + let db_size = match std::fs::metadata(&db_path) { + Ok(meta) => meta.len(), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0, + Err(e) => { + return Err(e).with_context(|| format!("checking size of {}", db_path.display())); + } + }; + + let wal_size = match std::fs::metadata(&wal_path) { + Ok(meta) => meta.len(), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0, + Err(e) => { + return Err(e).with_context(|| format!("checking size of {}", wal_path.display())); + } + }; + + Ok(StorageBytes::from_bytes(db_size + wal_size)) + } + + /// Return the total number of rows in the `trades` table. + /// + /// In DuckDB this is a cheap metadata operation on row-group + /// headers — safe to call frequently. + pub fn count_trades(&self) -> Result { + let conn = self.connection()?; + conn.query_row("SELECT COUNT(*) FROM trades", [], |r| r.get(0)) + .context("counting trades") + } + + /// Delete the `batch_size` oldest trades (by `ts`) and return the + /// number of rows actually deleted. + pub fn delete_oldest_trades_batch(&self, batch_size: i64) -> Result { + let conn = self.connection()?; + let deleted = conn + .execute( + "DELETE FROM trades WHERE rowid IN (\ + SELECT rowid FROM trades ORDER BY ts ASC LIMIT ?\ + )", + duckdb::params![batch_size], + ) + .context("deleting oldest trades batch")?; + Ok(deleted as u64) + } + + /// Rewrite the database file to reclaim filesystem space freed by + /// prior `DELETE` operations. `CHECKPOINT` alone only merges the + /// WAL — it does not shrink the main file. `VACUUM` is O(n) in + /// remaining rows, so callers should gate it behind a threshold. + /// + /// `VACUUM` requires exclusive table access, so it can fail with a + /// transaction conflict if the batch flusher is mid-append. This + /// method retries a few times with short sleeps — the flusher's + /// appender is only held open for a few milliseconds per flush, so + /// a brief wait is almost always enough. Safe to call from a + /// blocking thread (which is where cleanup always runs). + pub fn vacuum(&self) -> Result<()> { + const MAX_RETRIES: u32 = 3; + const RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); + + let mut last_err = None; + for attempt in 0..MAX_RETRIES { + let conn = self.connection()?; + match conn.execute_batch("VACUUM;") { + Ok(()) => return Ok(()), + Err(e) => { + last_err = Some(e); + if attempt + 1 < MAX_RETRIES { + tracing::debug!( + "VACUUM attempt {}/{} failed (likely batch flusher \ + holding appender); retrying in {RETRY_DELAY:?}", + attempt + 1, + MAX_RETRIES, + ); + std::thread::sleep(RETRY_DELAY); + } + } + } + } + // SAFETY: the loop always sets `last_err` before reaching this point. + Err(last_err.unwrap()) + .with_context(|| format!("vacuuming DuckDB database after {MAX_RETRIES} attempts")) + } + + /// Merge the DuckDB WAL into the main database file, then truncate + /// the WAL. This prevents the `.wal` file from doubling the on-disk + /// footprint after a bulk delete. + pub fn run_checkpoint(&self) -> Result<()> { + let conn = self.connection()?; + conn.execute_batch("CHECKPOINT;") + .context("checkpointing DuckDB WAL")?; + Ok(()) + } + /// Open a dedicated connection for the batch writer (shares the /// underlying `duckdb_database`). pub fn open_writer(&self) -> Result { @@ -364,11 +465,24 @@ impl Storage { } } + pub(crate) fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 + } + /// Delete every trade row whose `ts` (milliseconds since epoch) is /// older than `retention_hours`. Returns the number of deleted rows. - pub fn purge_old_trades(&self, retention_hours: u64) -> Result { + /// + /// When `retention_hours` is `0` (unlimited) the purge is skipped + /// and `Ok(0)` is returned immediately. + pub fn purge_old_trades(&self, retention_hours: RetentionHours) -> Result { + if retention_hours.as_hours() == 0 { + return Ok(0); + } let conn = self.connection()?; - let cutoff_ms = Self::now_ms() - (retention_hours as i64 * 3_600_000); + let cutoff_ms = Self::now_ms() - retention_hours.as_millis(); let deleted = conn .execute( "DELETE FROM trades WHERE ts < ?1", @@ -378,106 +492,10 @@ impl Storage { Ok(deleted as u64) } - /// Convenience: return the stored `last_cleanup` timestamp (milliseconds - /// since epoch), or `None` if cleanup has never run. - pub fn last_cleanup_ms(&self) -> Result> { - self.get_metadata("last_cleanup")? - .map(|v| v.parse::().context("parsing last_cleanup metadata")) - .transpose() - } - - pub fn record_cleanup(&self) -> Result<()> { + pub fn record_cleanup(&self) -> Result { let now_ms = Self::now_ms(); - self.set_metadata("last_cleanup", &now_ms.to_string()) - } - - /// Run a single data-retention cleanup pass. - /// - /// Deletes trades older than `retention_hours` and records the - /// `last_cleanup` timestamp so callers can avoid running it again too soon. - pub fn run_cleanup(&self, retention_hours: u64) { - match self.purge_old_trades(retention_hours) { - Ok(n) => { - if n > 0 { - tracing::info!("Cleaned up {n} trade(s) older than {retention_hours}h"); - } - if let Err(e) = self.record_cleanup() { - tracing::warn!("Failed to record last_cleanup timestamp: {e:#}"); - } - } - Err(e) => { - tracing::error!("Data cleanup failed: {e:#}"); - } - } - } - - /// Compute how long to sleep before the next cleanup is needed. - fn next_cleanup_delay(&self, retention_ms: i64) -> Duration { - let now_ms = Self::now_ms(); - - let anchor_ms = match self.last_cleanup_ms() { - Ok(Some(ts)) => ts, - Ok(None) => { - tracing::debug!("No last_cleanup recorded; anchoring at now"); - now_ms - } - Err(e) => { - tracing::warn!("Failed to read last_cleanup: {e:#}; retrying in 10 min"); - return Duration::from_secs(600); - } - }; - - let next_ms = anchor_ms + retention_ms; - if next_ms <= now_ms { - Duration::ZERO - } else { - Duration::from_millis((next_ms - now_ms) as u64) - } - } - - /// Spawn a background task that schedules the next cleanup pass based - /// on the `last_cleanup` metadata, without polling. - /// - /// After each cleanup pass (which writes `last_cleanup`), the task - /// computes `last_cleanup + retention_hours` and sleeps exactly until - /// that moment. This means wakeups only happen when data is actually - /// due for expiry — there is no periodic polling. - /// - /// A startup [`run_cleanup`](Self::run_cleanup) is expected to have been - /// called by the caller before this task is spawned so that `last_cleanup` - /// is initialised. - pub fn spawn_periodic_cleanup( - &self, - retention_hours: u64, - shutdown: CancellationToken, - ) -> JoinHandle<()> { - let storage = self.clone(); - tokio::spawn(async move { - let retention_ms = (retention_hours as i64) * 3_600_000; - - loop { - let delay = storage.next_cleanup_delay(retention_ms); - - tokio::select! { - biased; - _ = shutdown.cancelled() => { - tracing::info!("Periodic cleanup shut down."); - break; - } - _ = tokio::time::sleep(delay) => { - storage.run_cleanup(retention_hours); - } - } - } - }) - } - - /// Return the current UTC timestamp in milliseconds since the Unix epoch. - fn now_ms() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64 + self.set_metadata("last_cleanup", &now_ms.to_string())?; + Ok(now_ms) } /// Spawn a background task that receives trades on `rx`, buffers them,