Module watcher 

Directory watcher for incremental ingest.

Producers coordinate with the watcher via a simple sentinel-file protocol:

1. Producer atomically writes the data file (e.g. batch-0001.csv). Usually this means writing to batch-0001.csv.tmp first and renaming.
2. Producer creates a zero-byte companion file batch-0001.csv.ready.
3. The watcher detects the .ready file, appends the paired data file to the target table, and deletes both files on success.
4. On failure, the watcher moves both files into a failed/ subdirectory and writes a <name>.error JSON file with the error details. The failed files are not retried — manual intervention is expected.

Security: TOCTOU and atomic file operations

There is an inherent TOCTOU (time-of-check-to-time-of-use) race between detecting the .ready sentinel and opening the data file for ingest. Producers must use atomic file operations to avoid this:

• Write data to a temporary file (e.g. batch.csv.tmp), then rename it to the final name (batch.csv). Do not write directly to the target.
• Never replace a data file with a symlink between writing and creating the .ready sentinel — the watcher resolves symlinks via canonicalize(), but the window between existence check and open cannot be fully eliminated without kernel-level file descriptors.
• On shared filesystems, ensure the rename is atomic (same mount point).

Only one table per watched directory is supported; ingest is always in append mode. File extensions decide the ingest path: .csv/.json go through the CSV ingest (JSON-lines not supported today), .parquet/.pq through the Parquet ingest, and .arrow/.ipc/.feather through the Arrow IPC ingest.

Concurrency model

Each watcher runs as a tokio task and checks out connections from a per-watcher hyperdb_api::pool::Pool of hyperdb_api::AsyncConnections. Up to max_concurrent ingests run in parallel; every file runs inside its own BEGIN / COMMIT on its own pooled connection, so the engine’s primary sync connection (used by query, execute, chart, etc.) is never contended or forced to wait on a slow file.

notify delivers events through its own std mpsc channel; we forward them into a tokio::sync::mpsc on a small helper thread so the tokio consumer can .recv().await naturally.

Structs

WatchOptions

Options for start_watching. Use the builder-free literal form — every field has a sensible default.

WatcherHandle

Owns the notify watcher, the async ingest task, and the connection pool for one watched directory.

WatcherRegistry

Registry of all active watchers, keyed by canonicalized directory path.

WatcherStats

Running counters for a watcher. Updated in place as the background task processes events.

Constants

DEFAULT_MAX_CONCURRENT

Default ceiling on parallel ingests per watcher. Chosen conservatively — each parallel ingest holds one open TCP connection to hyperd plus a transaction, and most workloads have fewer than 4 incoming streams at a time.

MAX_CONCURRENT_LIMIT

Hard upper bound on max_concurrent to prevent a runaway watch_directory call from exhausting hyperd connections.

READY_SUFFIX

Suffix used for the sentinel (“ready”) file. Not a leading dot — append this to the full data file name (e.g. orders.csv → orders.csv.ready).

Functions

start_watching

Begin watching dir. Builds a dedicated connection pool, runs the initial sweep (sequentially — there’s no benefit to parallelism for startup since the pool isn’t under load yet), then installs a notify watcher that streams events to an async tokio task. Returns a snapshot of the initial-sweep stats.

stop_watching

Stop watching a directory. Returns a JSON summary including final stats.