feat: host several data store shards on one store

CMakeLists.txt

@@ -151,6 +151,8 @@ set(ELOQ_STORE_SOURCES
         src/tasks/write_buffer_aggregator.cpp
         src/tasks/prewarm_task.cpp
         src/tasks/read_task.cpp
+        src/tasks/attach_shard_task.cpp
+        src/tasks/detach_shard_task.cpp
         src/tasks/reopen_task.cpp
         src/tasks/scan_task.cpp
         src/tasks/task.cpp

docs/architecture/02-runtime-and-lifecycle.md

@@ -33,17 +33,124 @@ There is also a process-global `inline EloqStore *eloq_store` and a
 thread-local `inline thread_local Shard *shard` (`include/tasks/task.h`).
 Task code reaches its shard, options, and io manager through these — which is
 why **one process hosts at most one running EloqStore** and why task code must
-run on a shard thread.
+run on a shard thread. One store is enough: it hosts every data store shard
+the process serves (next sections).
 
 ## Status state machine
 
 `EloqStore::Status`: `Stopped → Starting → Running → Stopping → Stopped`.
-`Start()` CAS-transitions Stopped→Starting and rejects concurrent or repeated
-starts. `SendRequest` accepts work only while `Running`.
-
-## Start() sequence
-
-`EloqStore::Start(branch, term, partition_group_id)` —
+`StartInstance()` CAS-transitions Stopped→Starting and rejects concurrent or
+repeated starts. `SendRequest` accepts work only while `Running`, and only
+for an attached data store shard (next section).
+
+## Data store shards
+
+The embedder's unit of ownership is a *data store shard*: a set of partitions
+hosted as one unit, with its own term (in EloqDB, node group g's data is data
+store shard g). One store hosts several at once, each attached and detached
+on its own while the others keep serving, so one process runs one store and
+one budget (buffer pool, fd limit, local cache space) however many node groups
+its node leads. Only cloud mode hosts more than one: data on a node's own
+disk has one owner.
+
+- **Table.** `ds_shards_[id]` holds each shard's state (`DsShardState`:
+  Detached, Attaching, Attached, Expired, Detaching) and term, read without a
+  lock on every request. `AttachShard`, `DetachShard` and `Stop` change it
+  under `ds_shard_mux_`, which keeps their multi-field changes consistent when
+  the embedder calls them concurrently (an attach's "no other shard is
+  hosted" check and its claim; a late attach against `Stop`'s reset).
+  `MarkShardExpired` flips one slot with a compare-and-swap. The mutex is
+  never held across the term-file wait or the detach's wait for the worker
+  shards: in module builds a second caller blocking on it would block a
+  host worker pthread that drives a worker shard. Ids are below
+  `kMaxDsShards` (1024).
+- **Requests name their shard** (`KvRequest::SetDsShard`). `SendRequest`
+  resolves a request naming none to the default shard (the one `Start`
+  attached, else the first attached) and admits the request while the shard
+  is Attached or Expired; any other state refuses it (`NotRunning`). An
+  Expired shard's request goes on to its worker shard, which fails it with
+  `ExpiredTerm`: completing it inside `SendRequest` would run its callback
+  inside the submit call, which the fan-out sweeps do not expect. The worker
+  shard checks again when it
+  dequeues the request (`Shard::AdmitRequest`), so a detach that started in
+  between still refuses it.
+- **A request has no term; its task does.** The sender cannot tell when,
+  during a request's life, the shard's term changes, so the request does not
+  carry one. A request still on its way to its worker shard when the shard is
+  detached (a submission not yet returned, or one not yet dequeued) is
+  refused if the shard is still detached when the worker shard gets to it,
+  and runs in the new term if the shard has been attached again by then. What
+  holds is that a task runs entirely in one term: a detach waits for the
+  shard's running tasks, and an attach needs the slot Detached, so the
+  shard's term never changes under a running task.
+- **Terms travel with the task.** When a task starts (`Shard::StartTask`) it
+  copies the request's shard id and reads the shard's term from the table
+  once (`KvTask::ds_shard_id_`, `KvTask::term_`). Every switch to a task
+  (`Shard::SwitchToTask`) records both on the worker shard beside the running
+  task (`Shard::running_ds_shard_id_`, `Shard::running_term_`, read through
+  `ThdDsShard()` / `ThdTerm()` like `ThdTask()`), and
+  `AsyncIoManager::ProcessTerm()` returns that term: no lookup by partition
+  anywhere. The internal requests a task queues for
+  its partition (compaction, file GC, local GC, TTL cleaning, the reopen
+  driver behind parked waiters) are stamped with the task's shard, so they
+  are gated and counted like external ones and run in the same term. The fd
+  syncs that a file eviction runs from another shard's task name a manifest
+  by the fd's own term, as data files always were.
+- **`AttachShard(id, term)`** (cloud mode): raises the shard's term file
+  `CURRENT_TERM_<branch>_<id>` with a compare-and-swap, fails with
+  `ExpiredTerm` if the file holds a newer term, marks the slot Attached,
+  starts the shard's archive schedule (below) and queues the shard's prewarm.
+  The swap is an ordinary task: `AttachShard` sends one `AttachShardRequest`
+  to worker shard `id % num_threads` (bypassing the admission gate, as the
+  slot is still Attaching) and waits for it with a bthread-aware wait. The
+  `AttachShardTask` takes the shard's term from the slot like any task
+  (`StartTask`) and runs `CloudStoreMgr::UpsertTermFile` through that worker's
+  object store: read the file and its ETag, then a conditional upload
+  (`If-Match` on the ETag, or `If-None-Match: *` when it is missing), reading
+  again when a concurrent writer wins (409/412, or 404 on an `If-Match`), up
+  to 10 attempts. No thread blocks on HTTP.
+  Each partition then takes over lazily on its first open: `GetManifest`
+  promotes its manifest to the shard's term, and a manifest of a newer term
+  fails that partition's request with `ExpiredTerm` and marks the whole shard
+  Expired (`Shard::OnTaskFinished`), since the term file is per shard.
+- **`DetachShard(id)`** marks the slot Detaching (the gate), ends the shard's
+  archive schedule, then sends one
+  `DetachShardRequest` to every worker shard and waits for all of them. On
+  each worker the request runs as a `DetachShardTask`: it fails the requests
+  queued for the shard's partitions (pending writes, parked reopen waiters,
+  delayed reopen drivers), blocks until no task runs on them (the task that
+  brings `active_tasks_by_shard_` to 0 resumes it), then `EvictDsShard` drops
+  their root metas and cached pages, closes their files and drops their local
+  manifests (cloud mode). It finds the partitions without a lookup table:
+  of every partition with a root meta (`RootMetaMgr`) or an open file
+  (`IouringMgr::PartitionFiles`) on the worker, it drops those
+  `KvOptions::partition_filter` gives the detaching shard. The filter must
+  still give them when the detach runs, so the embedder detaches a shard
+  before reassigning its partitions; a leader change does not reassign
+  them. Hosting more than one shard needs the filter; `AttachShard` refuses
+  a second shard without it. The cached copies of data files stay:
+  a later term starts a new file, so no file of an older term changes, and a
+  later attach reuses them.
+- **Archives are scheduled per shard.** With scheduled archiving on
+  (`archive_interval_secs`), `ArchiveCrond` keeps one schedule per attached
+  shard: `AttachShard` starts it, first due one interval later, and
+  `DetachShard` ends it before draining, so a shard attached again at a newer
+  term starts afresh. One thread waits for the earliest due shard, submits
+  that shard's `GlobalArchiveRequest` (named with the shard's term, limited to
+  its partitions by `partition_filter`) with a completion callback, and sets
+  the shard's next due time one interval after the archive completes. The
+  archives of different shards run independently; an Expired or Detaching
+  shard's turn is put off one interval. File GC keeps
+  `num_retained_archives` per partition, so retention is per shard too.
+- **`Stop()`** drops everything wholesale and resets the table.
+- **`Start(branch, term, id)`** is `StartInstance(branch)` followed by
+  `AttachShard(id, term)`, for embedders that host one shard; a failed attach
+  stops the store again. `Term()` and `PartitionGroupId()` report the default
+  shard's term and id.
+
+## StartInstance() sequence
+
+`EloqStore::StartInstance(branch)` —
 the order is load-bearing; later steps assume earlier ones:
 
 1. `ValidateOptions(options_)` — sanity checks plus *rewrites* (e.g. in cloud
@@ -54,23 +161,26 @@ the order is load-bearing; later steps assume earlier ones:
 3. `InitStoreSpace()` — create/open every `store_path` root directory, hold
    `root_fds_`, build `store_path_lut` (partition→path placement honoring disk
    capacity or explicit weights).
-4. Record identity: `term_` (forced 0 in Local mode), `branch_`,
-   `partition_group_id_` (cloud only).
+4. Record `branch_`, which every hosted data store shard shares. Terms are
+   per shard (previous section).
 5. Compute per-shard fd budget: `fd_limit` clamped to process
    `RLIMIT_NOFILE` headroom minus `num_reserved_fd` (100), divided by
    `num_threads`. Fails with `OpenFileLimit` if insufficient.
 6. Construct + `Init()` every `Shard` (allocates queues; the heavy I/O
    bootstrap happens later on the shard's own thread).
-7. Start remote service: `CloudStorageService::Start()` then **block** on its
-   bootstrap (verifies bucket, CAS-upserts the `CURRENT_TERM_<branch>_<pg_id>`
-   object; a stale term fails the whole `Start`). Or `StandbyService::Start()`.
+7. Start remote service: `CloudStorageService::Start()` or
+   `StandbyService::Start()`. No term file is touched here: each
+   `AttachShard` raises its shard's.
 8. `Shard::Start()` for each shard — spawns worker threads (normal build) or
    marks shards runnable (module build).
 9. Module build only: register `EloqStoreModule` and notify workers.
 10. Start `ArchiveCrond` if `data_append_mode && num_retained_archives > 0 &&
-    archive_interval_secs > 0`.
-11. Start `PrewarmService` if `prewarm_cloud_cache` and mode is Cloud.
-12. Publish `Status::Running`.
+    archive_interval_secs > 0`; it archives each attached shard on that
+    shard's own schedule (see "Data store shards").
+11. Start `PrewarmService` if `prewarm_cloud_cache` and mode is Cloud; it
+    warms a shard's partitions when the shard is attached.
+12. Publish `Status::Running`. No request is admitted until a shard is
+    attached.
 
 Any failure routes through `fail_start` → `CleanupRuntime(started_shards)`, so
 partial startups unwind in the same order as full shutdown.
@@ -117,8 +227,9 @@ state:
   `global_registered_memories` XOR `pinned_memory_chunks`,
   `gc_global_mem_size_per_shard`, `pinned_tail_scratch_slots`,
   `segments_per_file_shift`.
-- `partition_filter` — predicate telling this instance which partitions it
-  owns (drives prewarm/global-op scoping in multi-instance deployments).
+- `partition_filter(ds_shard_id, tbl_id)` — predicate telling which
+  partitions belong to a hosted data store shard; scopes prewarm and the
+  global archive, reopen and create-branch sweeps to the shard they run for.
 - `LoadFromIni(path)` parses the same fields from an `.ini` file (used by
   benchmarks/tools; see `benchmark/*.ini`).
 

docs/architecture/03-request-api.md

@@ -99,4 +99,19 @@ Two retry loops live *inside* the engine (`Shard::StartTask` epilogue +
   up to `auto_oom_retry_times`, giving other tasks a chance to release pins.
 
 `ExpiredTerm` is terminal by design: it means another process owns a newer
-term and this writer must stop (doc 06).
+term and this writer must stop (doc 06). It is scoped to a data store shard:
+the shard whose partition met the newer term is marked Expired, every later
+request for it fails with `ExpiredTerm` on its worker shard, and the other
+shards the store hosts keep serving (doc 02).
+
+## Data store shard of a request
+
+Every request works for one data store shard (`KvRequest::SetDsShard`), the
+embedder's unit of ownership with its own term (doc 02). The shard decides
+whether the request is admitted and which term its partition's files are
+named with. A request naming no shard works for the store's default one: the
+shard `Start` attached, so embedders hosting one shard never set it. The
+global sweeps (`GlobalArchiveRequest`, `GlobalReopenRequest`,
+`GlobalCreateBranchRequest`, `DropTableRequest`) hand their shard to the
+per-partition requests they fan out, and keep only the partitions
+`partition_filter` gives that shard.

docs/architecture/06-persistence-and-recovery.md

@@ -96,10 +96,12 @@ the manifest and directory FDs.
 
 ## Terms
 
-The embedding system passes a `term` to `Start()` (0 in Local mode). Files and
-manifests are stamped with the writer's term. In cloud mode,
-`CloudStorageService` bootstrap CAS-updates `CURRENT_TERM_<branch>_<pg_id>`
-(ETag-guarded); a writer that discovers a newer published term gets
+The embedding system passes a `term` with each data store shard it attaches
+(`AttachShard`; `Start()` attaches one), 0 in Local mode. Files and
+manifests are stamped with the writer's term. In cloud mode the attach
+CAS-updates that shard's `CURRENT_TERM_<branch>_<pg_id>` (ETag-guarded), from
+a task on a worker shard (`AttachShardTask` → `CloudStoreMgr::UpsertTermFile`,
+doc 02); a writer that discovers a newer published term gets
 `KvError::ExpiredTerm` and must stop — this fences zombie writers after
 failover. Readers/GC treat *newer-term* files conservatively (see BranchGuard
 below). `ManifestTermFromFilename` etc. parse terms back out of names.

docs/architecture/07-io-stack.md

@@ -89,7 +89,8 @@ Local disk = bounded LRU cache of cloud objects; object store = durable copy.
 - **Concurrency limits**: cloud ops take a slot (`AcquireCloudSlot`,
   `max_cloud_concurrency`) and a pooled `DirectIoBuffer`
   (`AcquireCloudBuffer`); both park the coroutine when exhausted.
-- **Terms**: `ReadTermFile`/bootstrap CAS on `CURRENT_TERM_*` (doc 06);
+- **Terms**: `ReadTermFile`, and `UpsertTermFile`, the attach-time CAS on
+  `CURRENT_TERM_*` run from a task (docs 02, 06);
   `process_term_`/`partition_group_id_` stamp uploads and validate manifests.
 
 ## Cloud transport
@@ -101,8 +102,7 @@ Three layers, crossing the thread boundary exactly once each way:
    `KvTask` and shard.
 2. **`CloudStorageService`** (per store) — `cloud_request_threads` worker
    threads, each running a curl multi loop; jobs from a given shard always go
-   to the same worker (curl handles are not thread-safe). Also runs the
-   bootstrap term-file CAS at Start.
+   to the same worker (curl handles are not thread-safe).
 3. **`AsyncHttpManager`** (per shard, driven by the service worker) — builds
    signed HTTP requests via the pluggable `CloudBackend`
    (`include/storage/cloud_backend.h`; AWS S3 / GCS variants selected by

docs/architecture/08-data-lifecycle.md

@@ -86,8 +86,12 @@ requests in each `PendingWriteQueue`.
 - **CreateArchive / CreateBranch / DeleteBranch** — manifest-level operations
   (doc 06) executed under the write lock for the partition.
 
-`ArchiveCrond` (its own thread) periodically submits archive requests for all
-owned partitions at `archive_interval_secs`, bounded by `max_archive_tasks`.
+`ArchiveCrond` (its own thread) archives each attached data store shard on
+that shard's own schedule: every `archive_interval_secs`, measured from the
+end of its previous archive, starting one interval after the attach and
+ending at its detach (doc 02, "Data store shards"). Each archive is one
+`GlobalArchiveRequest` for the shard's partitions, bounded by
+`max_archive_tasks`.
 
 ## File garbage collection (`include/file_gc.h`, `src/file_gc.cpp`)
 

docs/architecture/10-bindings-and-embedding.md

@@ -65,6 +65,17 @@ only when the embedding runtime (EloqKV's tx service, with bthread) defines
 - Shutdown is cooperative: shards are flagged `Stopping` and the host's
   workers run rounds until each reports `Stopped` (see
   `EloqStore::CleanupRuntime`).
+- A host that leads several node groups hosts their data store shards on the
+  one store: `StartInstance` once, then `AttachShard`/`DetachShard` as
+  leadership comes and goes, each request stamped with its shard
+  (`KvRequest::SetDsShard`) and `partition_filter(ds_shard_id, tbl_id)`
+  answering from the host's bucket map (doc 02); the filter is required to
+  host more than one shard, since a detach drops the state of the
+  partitions it gives the detaching shard, and `DetachShard` must run before
+  the bucket map moves those partitions elsewhere. `AttachShard` and
+  `DetachShard` wait with bthread-aware waits, so an RPC handler's bthread
+  may call them; a module hook (`Process`) must not, since `DetachShard`
+  waits for every worker shard, its own included.
 
 When changing `Shard::WorkLoop`, mirror the change in `WorkOneRound` — they
 are the same scheduler in two drive modes.