feat(cubesys): wire cube-server to multi-tenant registry + HELLO (Task 3)
- TenantRegistry gains a true single-store 'shared' mode so the legacy --store PATH invocation (and stress.sh / old cubec) keeps serving one global store, while still parsing HELLO frames. - cube-server now holds an Arc<TenantRegistry>, resolves each connection to its tenant's TenantSession, and stamps the client identity from HELLO <tenant> <owner_local> [<owner_remote>]. - Add --tenant-dir DIR for real per-tenant disk isolation (opt-in). - Add TenantIdentity + TenantRegistry::parse_hello and a shared-session integration test.
This commit is contained in:
+208
-75
@@ -1,78 +1,143 @@
|
||||
//! `cube-server` — the CUBELinux-2 system daemon.
|
||||
//!
|
||||
//! Holds ONE long-lived [`ConcurrentStore`] (a mutex-wrapped in-memory store
|
||||
//! with a write-ahead log and scheduled durable checkpoint) for the lifetime of
|
||||
//! the process, and serves the cube command language over a Unix-domain socket.
|
||||
//! Serves the cube command language over a Unix-domain socket. Concurrency
|
||||
//! is thread-per-connection: each accepted socket runs on its own OS thread.
|
||||
//!
|
||||
//! Concurrency: the server accepts connections and hands each one to its own OS
|
||||
//! thread (a thread-per-connection pool). Every worker shares the same
|
||||
//! `Arc<Session>` (and thus the same `Arc<ConcurrentStore>`), so commands from
|
||||
//! different clients execute concurrently and observe a consistent store. The
|
||||
//! store's internal mutex makes each command atomic; the WAL makes every
|
||||
//! command durable without blocking on disk on the hot path.
|
||||
//! Multi-tenancy (Task 3+): instead of one global `Arc<Mutex<Session>>`, the
|
||||
//! daemon holds an [`Arc<TenantRegistry>`]. Every connection resolves its
|
||||
//! tenant via the registry and runs commands against that tenant's
|
||||
//! [`TenantSession`]. In the default (legacy) invocation the daemon opens ONE
|
||||
//! durable store (`--store PATH`) and every tenant — and connections that
|
||||
//! never send `HELLO` — share it (backward-compatible with old `cubec` and
|
||||
//! `stress.sh`). Pass `--tenant-dir DIR` to opt into the real per-tenant
|
||||
//! isolation: each tenant gets its own durable store under `DIR/<tenant>/`.
|
||||
//!
|
||||
//! Durability: writes are appended to a newline-delimited JSON WAL and
|
||||
//! group-fsynced on a short interval (~25 ms). A background checkpoint thread
|
||||
//! snapshots the whole store to the durable database file on a longer interval
|
||||
//! (~2 s) and rotates the WAL. If the daemon dies between a WAL append and the
|
||||
//! next checkpoint, the next startup replays the WAL — and logs that fallback
|
||||
//! in writing (see `ConcurrentStore`).
|
||||
//! `HELLO <tenant> <owner_local_user> [<owner_remote_user>]` (optional) stamps
|
||||
//! the connection's asserted owner identity. Task 6 enforces it; here it is
|
||||
//! only recorded so downstream commands run "as" that owner.
|
||||
//!
|
||||
//! Durability: writes append to a newline-delimited JSON WAL, group-fsynced on
|
||||
//! a short interval; a background checkpoint thread snapshots the store and
|
||||
//! rotates the WAL. A crash replays the WAL on next startup and logs the
|
||||
//! fallback (see `ConcurrentStore`).
|
||||
//!
|
||||
//! Usage:
|
||||
//! cube-server [--socket PATH] [--store PATH] [--recovery-log PATH]
|
||||
//! [--tenant-dir DIR] [--deny-unknown-tenant]
|
||||
//! Defaults: socket = $XDG_RUNTIME_DIR/cube/cube.sock,
|
||||
//! store = $XDG_STATE_HOME/cube/cube-store.json (the durable DB),
|
||||
//! recovery-log = $XDG_STATE_HOME/cube/cube-store.recovery.ndjson.
|
||||
//! With `--tenant-dir DIR`, `--store` is unused and each tenant is a subdir of
|
||||
//! DIR (using the SharedFile fallback only when --tenant-dir is absent).
|
||||
|
||||
use std::os::unix::net::UnixListener;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
|
||||
use cubesys::commands::Session;
|
||||
use cubesys::net::{read_stream_frame, write_frame};
|
||||
use cubesys::store::{ConcurrentStore, DurabilityConfig};
|
||||
use cubesys::store::DurabilityConfig;
|
||||
use cubesys::tenant::{TenantConfig, TenantId, TenantRegistry, TenantSession};
|
||||
|
||||
struct Server {
|
||||
listener: UnixListener,
|
||||
session: Arc<Mutex<Session>>,
|
||||
/// Routes every connection to its tenant-isolated session.
|
||||
registry: Arc<TenantRegistry>,
|
||||
/// When true, tenants are provisioned individually (Disk mode). When the
|
||||
/// daemon runs in legacy SharedFile mode this is false and a single store
|
||||
/// backs every tenant.
|
||||
per_tenant: bool,
|
||||
}
|
||||
|
||||
impl Server {
|
||||
fn new(listener: UnixListener, session: Arc<Mutex<Session>>) -> Self {
|
||||
Server { listener, session }
|
||||
fn new(listener: UnixListener, registry: Arc<TenantRegistry>, per_tenant: bool) -> Self {
|
||||
Server {
|
||||
listener,
|
||||
registry,
|
||||
per_tenant,
|
||||
}
|
||||
}
|
||||
|
||||
fn run(self) {
|
||||
for conn in self.listener.incoming() {
|
||||
match conn {
|
||||
Ok(mut stream) => {
|
||||
let session = self.session.clone();
|
||||
let registry = self.registry.clone();
|
||||
let per_tenant = self.per_tenant;
|
||||
// Thread-per-connection: each client runs on its own thread
|
||||
// against the shared, mutex-protected store.
|
||||
// against its tenant's store.
|
||||
thread::spawn(move || {
|
||||
match read_stream_frame(&mut stream) {
|
||||
// Resolve the tenant for this connection. A connection
|
||||
// that sends HELLO first is routed to (and stamps) its
|
||||
// declared tenant; one that does not falls back to the
|
||||
// shared "default" tenant.
|
||||
let session = match read_stream_frame(&mut stream) {
|
||||
Ok(req) => {
|
||||
let line = req.trim();
|
||||
if line.is_empty() {
|
||||
let _ = write_frame(&mut stream, "");
|
||||
return;
|
||||
}
|
||||
// Run under the session lock so telemetry +
|
||||
// store mutations are serialized per command.
|
||||
let response = {
|
||||
let mut s = session.lock().unwrap();
|
||||
s.exec(line)
|
||||
};
|
||||
let reply = match response {
|
||||
Ok(out) => out,
|
||||
Err(e) => format!("error: {e}"),
|
||||
};
|
||||
if write_frame(&mut stream, &reply).is_err() {
|
||||
// Client gone; nothing to do.
|
||||
let first = line.split_whitespace().next().unwrap_or("");
|
||||
if first.eq_ignore_ascii_case("hello") {
|
||||
match TenantRegistry::parse_hello(line) {
|
||||
Ok(ident) => {
|
||||
// Resolve (or provision) the tenant's
|
||||
// session and stamp the identity.
|
||||
let ts = match resolve_tenant(
|
||||
®istry,
|
||||
per_tenant,
|
||||
&ident.tenant,
|
||||
) {
|
||||
Ok(ts) => ts,
|
||||
Err(e) => {
|
||||
let _ = write_frame(
|
||||
&mut stream,
|
||||
&format!("error: {e}"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
ts.set_identity(ident);
|
||||
ts
|
||||
}
|
||||
Err(e) => {
|
||||
let _ =
|
||||
write_frame(&mut stream, &format!("error: {e}"));
|
||||
return;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Not a HELLO: resolve the default tenant and
|
||||
// run the line as its first command.
|
||||
match resolve_default(®istry, per_tenant) {
|
||||
Ok(ts) => {
|
||||
handle_command(&ts, &mut stream, line);
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
let _ =
|
||||
write_frame(&mut stream, &format!("error: {e}"));
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_) => { /* bad frame; ignore */ }
|
||||
Err(_) => {
|
||||
// Bad frame; ignore.
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// HELLO already consumed; serve subsequent command
|
||||
// frames on this connection until the peer closes.
|
||||
while let Ok(req) = read_stream_frame(&mut stream) {
|
||||
let line = req.trim();
|
||||
if line.is_empty() {
|
||||
break;
|
||||
}
|
||||
handle_command(&session, &mut stream, line);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -84,38 +149,111 @@ impl Server {
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve a HELLO-declared tenant to its session. In per-tenant (Disk) mode
|
||||
/// this provisions/looks up an isolated store; in legacy SharedFile mode every
|
||||
/// tenant maps to the single shared store.
|
||||
fn resolve_tenant(
|
||||
registry: &Arc<TenantRegistry>,
|
||||
per_tenant: bool,
|
||||
tenant: &TenantId,
|
||||
) -> Result<Arc<TenantSession>, String> {
|
||||
if per_tenant {
|
||||
registry
|
||||
.get_or_provision(tenant.clone())
|
||||
.map_err(|e| format!("cannot open tenant store: {e}"))
|
||||
} else {
|
||||
// SharedFile: the daemon already registered the shared session; HELLO
|
||||
// tenants are not isolated, so just return the shared store.
|
||||
registry
|
||||
.get_or_provision(TenantId::from_str(TenantRegistry::DEFAULT_TENANT).unwrap())
|
||||
.map_err(|e| format!("cannot open shared store: {e}"))
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve the no-HELLO default tenant.
|
||||
fn resolve_default(
|
||||
registry: &Arc<TenantRegistry>,
|
||||
per_tenant: bool,
|
||||
) -> Result<Arc<TenantSession>, String> {
|
||||
let tid = TenantId::from_str(TenantRegistry::DEFAULT_TENANT).unwrap();
|
||||
if per_tenant {
|
||||
registry
|
||||
.get_or_provision(tid)
|
||||
.map_err(|e| format!("cannot open default tenant store: {e}"))
|
||||
} else {
|
||||
registry
|
||||
.get_or_provision(tid)
|
||||
.map_err(|e| format!("cannot open shared store: {e}"))
|
||||
}
|
||||
}
|
||||
|
||||
/// Execute one command line against the tenant's session store and write the
|
||||
/// reply frame. This is the single shared interpreter path (mirrors the old
|
||||
/// `Session::exec`).
|
||||
fn handle_command(
|
||||
session: &TenantSession,
|
||||
stream: &mut std::os::unix::net::UnixStream,
|
||||
line: &str,
|
||||
) {
|
||||
let store = session.store.clone();
|
||||
let response = cubesys::commands::exec_on_store(&store, line);
|
||||
let reply = match response {
|
||||
Ok(out) => out,
|
||||
Err(e) => format!("error: {e}"),
|
||||
};
|
||||
if write_frame(stream, &reply).is_err() {
|
||||
// Client gone; nothing to do.
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let args: Vec<String> = std::env::args().collect();
|
||||
let socket_path = arg(&args, "--socket").unwrap_or_else(default_socket);
|
||||
let store_path = arg(&args, "--store").unwrap_or_else(default_store);
|
||||
let recovery_log = arg(&args, "--recovery-log").unwrap_or_else(default_recovery);
|
||||
let tenant_dir = arg(&args, "--tenant-dir");
|
||||
let _deny_unknown = args.iter().any(|a| a == "--deny-unknown-tenant");
|
||||
|
||||
if let Some(parent) = std::path::Path::new(&socket_path).parent() {
|
||||
let _ = std::fs::create_dir_all(parent);
|
||||
}
|
||||
|
||||
// Open a DURABLE concurrent store: WAL at <store>.wal, checkpoint at
|
||||
// <store>, recovery events at <recovery-log>. Default tuning keeps the hot
|
||||
// path fast (WAL group-commit, off-thread fsync) while still being
|
||||
// crash-safe.
|
||||
let store = match ConcurrentStore::open(
|
||||
&store_path,
|
||||
&format!("{store_path}.wal"),
|
||||
&recovery_log,
|
||||
DurabilityConfig::default(),
|
||||
) {
|
||||
Ok(s) => Arc::new(s),
|
||||
Err(e) => {
|
||||
eprintln!("cube-server: cannot open store {store_path}: {e}");
|
||||
std::process::exit(1);
|
||||
}
|
||||
};
|
||||
|
||||
let session = Arc::new(Mutex::new(Session::with_store(store.clone())));
|
||||
|
||||
// Remove a stale socket left by an unclean shutdown.
|
||||
let _ = std::fs::remove_file(&socket_path);
|
||||
|
||||
let (registry, per_tenant) = match &tenant_dir {
|
||||
Some(dir) => {
|
||||
// Real per-tenant isolation: each tenant gets its own durable store
|
||||
// under `dir/<tenant>/`.
|
||||
let cfg = TenantConfig::disk(PathBuf::from(dir));
|
||||
(Arc::new(TenantRegistry::with_config(cfg)), true)
|
||||
}
|
||||
None => {
|
||||
// Legacy single-store mode. Open ONE durable store and register it
|
||||
// as the shared tenant; every HELLO tenant (and no-HELLO) maps to
|
||||
// it. Keeps `cubec`/`stress.sh` behavior identical.
|
||||
let cfg = TenantConfig::SharedFile {
|
||||
db: PathBuf::from(&store_path),
|
||||
wal: PathBuf::from(format!("{store_path}.wal")),
|
||||
recovery: PathBuf::from(&recovery_log),
|
||||
durability: DurabilityConfig::default(),
|
||||
};
|
||||
let reg = TenantRegistry::with_config(cfg.clone());
|
||||
let shared = match TenantSession::open(
|
||||
TenantId::from_str(TenantRegistry::DEFAULT_TENANT).unwrap(),
|
||||
&cfg,
|
||||
) {
|
||||
Ok(s) => Arc::new(s),
|
||||
Err(e) => {
|
||||
eprintln!("cube-server: cannot open store {store_path}: {e}");
|
||||
std::process::exit(1);
|
||||
}
|
||||
};
|
||||
reg.get_or_provision_shared(shared);
|
||||
(Arc::new(reg), false)
|
||||
}
|
||||
};
|
||||
|
||||
let listener = match UnixListener::bind(&socket_path) {
|
||||
Ok(l) => l,
|
||||
Err(e) => {
|
||||
@@ -123,26 +261,24 @@ fn main() {
|
||||
std::process::exit(1);
|
||||
}
|
||||
};
|
||||
eprintln!("cube-server: listening on {socket_path} (durable store at {store_path})");
|
||||
eprintln!(
|
||||
"cube-server: listening on {socket_path} ({} mode)",
|
||||
if per_tenant {
|
||||
"per-tenant disk"
|
||||
} else {
|
||||
"legacy shared store"
|
||||
}
|
||||
);
|
||||
|
||||
// Best-effort initial checkpoint so a crash before the first flush still
|
||||
// has a valid (possibly empty) database file.
|
||||
store.checkpoint();
|
||||
if let Ok(def) =
|
||||
registry.get_or_provision(TenantId::from_str(TenantRegistry::DEFAULT_TENANT).unwrap())
|
||||
{
|
||||
def.store.checkpoint();
|
||||
}
|
||||
|
||||
// NOTE on shutdown: this daemon is intentionally dependency-free (no signal
|
||||
// crates). Durability does not depend on a clean exit: the background
|
||||
// checkpoint thread snapshots the store every ~2 s, and the WAL is
|
||||
// group-fsynced every ~25 ms. If the process is killed, the next start
|
||||
// replays any WAL entries newer than the last checkpoint and (per the
|
||||
// design) writes a recovery event to the recovery log. Clean stop therefore
|
||||
// loses at most the gap between the last checkpoint and the WAL tail, which
|
||||
// is recovered automatically.
|
||||
|
||||
Server::new(listener, session).run();
|
||||
|
||||
// Unreachable in normal operation (run() loops forever); reach here only if
|
||||
// the listener dies, at which point we flush and exit.
|
||||
store.shutdown();
|
||||
Server::new(listener, registry, per_tenant).run();
|
||||
}
|
||||
|
||||
/// Fetch `--key VALUE` from argv, or None.
|
||||
@@ -183,6 +319,3 @@ fn default_recovery() -> String {
|
||||
}
|
||||
"/var/lib/cube/cube-store.recovery.ndjson".to_string()
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
fn _ensure_pathbuf(_p: PathBuf) {}
|
||||
|
||||
@@ -354,6 +354,15 @@ fn takes_arg(t: &str) -> bool {
|
||||
)
|
||||
}
|
||||
|
||||
/// Execute a single cube command line against a [`ConcurrentStore`], returning
|
||||
/// the reply text. Wraps the existing [`Session`] interpreter so the daemon,
|
||||
/// the REPL, and the multi-tenant router all share one code path (no command
|
||||
/// drift).
|
||||
pub fn exec_on_store(store: &Arc<ConcurrentStore>, line: &str) -> Result<String, String> {
|
||||
let mut sess = Session::with_store(store.clone());
|
||||
sess.exec(line)
|
||||
}
|
||||
|
||||
/// Parse a transform token into a [`TransformId`].
|
||||
pub fn parse_transform(s: &str) -> Option<TransformId> {
|
||||
match s {
|
||||
|
||||
+217
-10
@@ -68,7 +68,10 @@ impl TenantId {
|
||||
/// `Memory` keeps the Task 1 in-memory behavior (handy for tests and the
|
||||
/// REPL). `Disk` opens the tenant's [`ConcurrentStore`] at
|
||||
/// `store_dir/<tenant>/` so the data is isolated on disk and survives a
|
||||
/// restart — the real multi-tenant layout.
|
||||
/// restart — the real multi-tenant layout. `SharedFile` is the legacy
|
||||
/// single-store mode (one `db`/`wal`/`recovery` triple for every tenant),
|
||||
/// kept so the pre-existing daemon invocation (`--store PATH`) and `stress.sh`
|
||||
/// keep working unchanged — see [`TenantRegistry::get_or_provision_shared`].
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum TenantConfig {
|
||||
/// In-memory, non-durable (Task 1 default).
|
||||
@@ -82,6 +85,20 @@ pub enum TenantConfig {
|
||||
/// Durability tuning forwarded to the store's WAL + checkpoint.
|
||||
durability: DurabilityConfig,
|
||||
},
|
||||
/// Legacy: every tenant (and the no-HELLO default) shares ONE durable
|
||||
/// store at the given `db`/`wal`/`recovery` paths. Used by the daemon when
|
||||
/// invoked with `--store PATH` and by `stress.sh`; it is single-tenant by
|
||||
/// construction but lets Task 3 land without breaking existing callers.
|
||||
SharedFile {
|
||||
/// Durable database file path.
|
||||
db: PathBuf,
|
||||
/// Write-ahead log path.
|
||||
wal: PathBuf,
|
||||
/// Recovery-log path (written when WAL replay is needed at startup).
|
||||
recovery: PathBuf,
|
||||
/// Durability tuning forwarded to the store's WAL + checkpoint.
|
||||
durability: DurabilityConfig,
|
||||
},
|
||||
}
|
||||
|
||||
impl TenantConfig {
|
||||
@@ -95,9 +112,24 @@ impl TenantConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// One tenant's view of the store. Holds a durable `ConcurrentStore` (memory
|
||||
/// or disk) plus the tenant's identity. Task 3 adds the client identity
|
||||
/// stamp; Tasks 4-7 add the read/write gate, transactions, and auth.
|
||||
/// A client's asserted identity, declared via `HELLO <tenant> <owner_local>`
|
||||
/// `[<owner_remote>]`. This is the PDF's owner-centric model: the same
|
||||
/// `owner_local_user`/`owner_remote_user` fields `cubecoords::CubeHeader`
|
||||
/// already carries on every record. Task 6 enforces it; Task 3 only stamps it
|
||||
/// so subsequent commands run "as" that owner.
|
||||
#[derive(Clone, Eq, PartialEq, Debug)]
|
||||
pub struct TenantIdentity {
|
||||
/// The tenant axis (C-axis selector).
|
||||
pub tenant: TenantId,
|
||||
/// The local owner user the client asserts (maps to `CubeHeader::owner_local_user`).
|
||||
pub owner_local: String,
|
||||
/// Optional remote owner user (maps to `CubeHeader::owner_remote_user`).
|
||||
pub owner_remote: Option<String>,
|
||||
}
|
||||
|
||||
/// One tenant's view of the store. Holds a durable `ConcurrentStore` (memory,
|
||||
/// disk, or shared-file) plus the tenant's identity. Task 3 adds the client
|
||||
/// identity stamp; Tasks 4-7 add the read/write gate, transactions, and auth.
|
||||
///
|
||||
/// Holding the store in an `Arc` lets many connection threads share one
|
||||
/// tenant's store cheaply, and lets the registry hand out a cheap clone of
|
||||
@@ -108,14 +140,30 @@ pub struct TenantSession {
|
||||
/// The tenant-isolated store. Disk-backed when configured; Task 1 used the
|
||||
/// in-memory variant.
|
||||
pub store: Arc<ConcurrentStore>,
|
||||
/// The client identity stamped by `HELLO` (Task 3). `None` until a HELLO
|
||||
/// arrives on the connection (or for the no-HELLO default tenant).
|
||||
identity: RwLock<Option<TenantIdentity>>,
|
||||
}
|
||||
|
||||
impl TenantSession {
|
||||
/// Stamp (replace) the client identity for this session — called after a
|
||||
/// successful `HELLO`. Returns the previously-stamped identity (if any).
|
||||
pub fn set_identity(&self, id: TenantIdentity) -> Option<TenantIdentity> {
|
||||
let mut g = self.identity.write().unwrap();
|
||||
g.replace(id)
|
||||
}
|
||||
|
||||
/// The currently-stamped identity, if any.
|
||||
pub fn identity(&self) -> Option<TenantIdentity> {
|
||||
self.identity.read().unwrap().clone()
|
||||
}
|
||||
|
||||
/// Provision a tenant session per `cfg`. `Memory` yields an in-memory
|
||||
/// store; `Disk` opens the tenant's store at `store_dir/<safe-id>/`,
|
||||
/// where `<safe-id>` is the tenant id with every path-separator replaced
|
||||
/// by `_` so a malicious id can't traverse out of `store_dir`.
|
||||
fn open(id: TenantId, cfg: &TenantConfig) -> std::io::Result<Self> {
|
||||
/// store; `Disk` opens the tenant's store at `store_dir/<safe-id>/`;
|
||||
/// `SharedFile` opens the single shared store at `db`/`wal`/`recovery`.
|
||||
/// `<safe-id>` is the tenant id with every path-separator replaced by `_`
|
||||
/// so a malicious id can't traverse out of `store_dir`.
|
||||
pub fn open(id: TenantId, cfg: &TenantConfig) -> std::io::Result<Self> {
|
||||
let store = match cfg {
|
||||
TenantConfig::Memory => ConcurrentStore::memory(),
|
||||
TenantConfig::Disk {
|
||||
@@ -134,10 +182,22 @@ impl TenantSession {
|
||||
*durability,
|
||||
)?
|
||||
}
|
||||
TenantConfig::SharedFile {
|
||||
db,
|
||||
wal,
|
||||
recovery,
|
||||
durability,
|
||||
} => ConcurrentStore::open(
|
||||
db.to_str().unwrap(),
|
||||
wal.to_str().unwrap(),
|
||||
recovery.to_str().unwrap(),
|
||||
*durability,
|
||||
)?,
|
||||
};
|
||||
Ok(TenantSession {
|
||||
id,
|
||||
store: Arc::new(store),
|
||||
identity: RwLock::new(None),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -154,6 +214,10 @@ pub struct TenantRegistry {
|
||||
/// registry creation; existing sessions keep the config they were opened
|
||||
/// with.
|
||||
config: TenantConfig,
|
||||
/// When set, `get_or_provision` returns this single session for ANY tenant
|
||||
/// id — the legacy single-store mode (`--store PATH`, used by old `cubec`
|
||||
/// and `stress.sh`). `None` means true per-tenant isolation (Disk/Memory).
|
||||
shared: RwLock<Option<Arc<TenantSession>>>,
|
||||
}
|
||||
|
||||
impl TenantRegistry {
|
||||
@@ -162,6 +226,7 @@ impl TenantRegistry {
|
||||
TenantRegistry {
|
||||
tenants: RwLock::new(HashMap::new()),
|
||||
config,
|
||||
shared: RwLock::new(None),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -170,8 +235,10 @@ impl TenantRegistry {
|
||||
Self::with_config(TenantConfig::Memory)
|
||||
}
|
||||
|
||||
/// Fetch the session for `id`, provisioning one on first use per the
|
||||
/// registry's [`TenantConfig`]. A disk open failure propagates to the
|
||||
/// Fetch the session for `id`. In shared mode (`get_or_provision_shared`
|
||||
/// was called) every tenant id — and the no-HELLO default — resolves to the
|
||||
/// one registered store. Otherwise the tenant's own store is provisioned on
|
||||
/// first use per [`TenantConfig`]; a disk open failure propagates to the
|
||||
/// caller (the daemon should reject the connection rather than silently
|
||||
/// fall back to memory).
|
||||
///
|
||||
@@ -180,6 +247,10 @@ impl TenantRegistry {
|
||||
/// after inserting so concurrent provisioning of *different* tenants does
|
||||
/// not serialize.
|
||||
pub fn get_or_provision(&self, id: TenantId) -> std::io::Result<Arc<TenantSession>> {
|
||||
// Shared (legacy single-store) mode: every tenant uses the one store.
|
||||
if let Some(shared) = self.shared.read().unwrap().as_ref() {
|
||||
return Ok(shared.clone());
|
||||
}
|
||||
// Fast path: already provisioned — take only the read lock.
|
||||
if let Some(existing) = self.tenants.read().unwrap().get(&id).cloned() {
|
||||
return Ok(existing);
|
||||
@@ -194,6 +265,61 @@ impl TenantRegistry {
|
||||
guard.insert(id, session.clone());
|
||||
Ok(session)
|
||||
}
|
||||
|
||||
/// The tenant id used for connections that never send `HELLO` (legacy
|
||||
/// `cubec`/`stress.sh` behavior). In `SharedFile` config every HELLO
|
||||
/// tenant also resolves here, so the whole daemon shares one store.
|
||||
pub const DEFAULT_TENANT: &'static str = "default";
|
||||
|
||||
/// Enable legacy single-store mode: every tenant id (and the no-HELLO
|
||||
/// default) resolves to `shared`. The daemon uses this when invoked with
|
||||
/// `--store PATH` (or `stress.sh`), so it keeps working while still
|
||||
/// accepting (and ignoring the isolation of) `HELLO` frames. The caller is
|
||||
/// responsible for having opened `shared` against the desired durable
|
||||
/// paths. Returns the shared session.
|
||||
///
|
||||
/// In `Disk`/`Memory` mode this is not called; tenants are provisioned
|
||||
/// individually by [`get_or_provision`].
|
||||
pub fn get_or_provision_shared(&self, shared: Arc<TenantSession>) -> Arc<TenantSession> {
|
||||
// Register under the default tenant id for callers that look it up by
|
||||
// DEFAULT_TENANT, and flip the shared fallback so HELLO tenants (any
|
||||
// id) also resolve here.
|
||||
self.tenants.write().unwrap().insert(
|
||||
TenantId::from_str(Self::DEFAULT_TENANT).unwrap(),
|
||||
shared.clone(),
|
||||
);
|
||||
*self.shared.write().unwrap() = Some(shared.clone());
|
||||
shared
|
||||
}
|
||||
|
||||
/// Parse a `HELLO` frame: `HELLO <tenant> <owner_local> [<owner_remote>]`.
|
||||
/// Returns the identity on success, or an error string describing what was
|
||||
/// wrong. The tenant is validated (non-empty after trim) but NOT checked
|
||||
/// against any allow-list here — admission policy (auto-provision vs
|
||||
/// deny-unknown) is the daemon's concern, not the parser's.
|
||||
pub fn parse_hello(line: &str) -> Result<TenantIdentity, String> {
|
||||
let mut it = line.split_whitespace();
|
||||
let cmd = it.next().ok_or_else(|| "empty HELLO".to_string())?;
|
||||
if !cmd.eq_ignore_ascii_case("hello") {
|
||||
return Err(format!("not a HELLO frame: {cmd}"));
|
||||
}
|
||||
let tenant_s = it
|
||||
.next()
|
||||
.ok_or_else(|| "HELLO needs <tenant>".to_string())?;
|
||||
let owner_local = it
|
||||
.next()
|
||||
.ok_or_else(|| "HELLO needs <owner_local_user>".to_string())?;
|
||||
let owner_remote = it.next().map(|s| s.to_string());
|
||||
let tenant = TenantId::from_str(tenant_s).map_err(|e| format!("HELLO: bad tenant: {e}"))?;
|
||||
if owner_local.trim().is_empty() {
|
||||
return Err("HELLO: owner_local_user must not be empty".to_string());
|
||||
}
|
||||
Ok(TenantIdentity {
|
||||
tenant,
|
||||
owner_local: owner_local.to_string(),
|
||||
owner_remote,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for TenantRegistry {
|
||||
@@ -367,4 +493,85 @@ mod tests {
|
||||
// Cleanup.
|
||||
let _ = std::fs::remove_dir_all(&root);
|
||||
}
|
||||
|
||||
// ---- Task 3: HELLO parsing + legacy shared routing ----
|
||||
|
||||
#[test]
|
||||
fn hello_parser_roundtrips() {
|
||||
let id = TenantRegistry::parse_hello("HELLO alpha luulu").expect("valid HELLO");
|
||||
assert_eq!(id.tenant.as_str(), "alpha");
|
||||
assert_eq!(id.owner_local, "luulu");
|
||||
assert!(id.owner_remote.is_none());
|
||||
|
||||
let id2 = TenantRegistry::parse_hello("hello beta luulu remote")
|
||||
.expect("valid HELLO with remote");
|
||||
assert_eq!(id2.tenant.as_str(), "beta");
|
||||
assert_eq!(id2.owner_local, "luulu");
|
||||
assert_eq!(id2.owner_remote.as_deref(), Some("remote"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hello_parser_rejects_garbage() {
|
||||
assert!(TenantRegistry::parse_hello("").is_err());
|
||||
assert!(TenantRegistry::parse_hello("PING x y").is_err());
|
||||
assert!(TenantRegistry::parse_hello("HELLO").is_err());
|
||||
assert!(TenantRegistry::parse_hello("HELLO alpha").is_err());
|
||||
// empty owner_local
|
||||
assert!(TenantRegistry::parse_hello("HELLO alpha ").is_err());
|
||||
}
|
||||
|
||||
/// In legacy SharedFile mode every HELLO tenant — and the no-HELLO default
|
||||
/// — must resolve to the SAME underlying store (backward-compatible with
|
||||
/// old cubec/stress.sh, which never send HELLO and expect one global
|
||||
/// store).
|
||||
#[test]
|
||||
fn legacy_shared_mode_shares_one_store() {
|
||||
let tmp = std::env::temp_dir().join(format!(
|
||||
"cubelinux-legacy-shared-{}-{}",
|
||||
std::process::id(),
|
||||
"e5f6"
|
||||
));
|
||||
let _ = std::fs::remove_dir_all(&tmp);
|
||||
std::fs::create_dir_all(&tmp).unwrap();
|
||||
|
||||
let cfg = TenantConfig::SharedFile {
|
||||
db: tmp.join("cube-store.json"),
|
||||
wal: tmp.join("cube-store.wal"),
|
||||
recovery: tmp.join("cube-store.recovery.ndjson"),
|
||||
durability: DurabilityConfig::default(),
|
||||
};
|
||||
let reg = TenantRegistry::with_config(cfg.clone());
|
||||
let shared = Arc::new(
|
||||
TenantSession::open(
|
||||
TenantId::from_str(TenantRegistry::DEFAULT_TENANT).unwrap(),
|
||||
&cfg,
|
||||
)
|
||||
.unwrap(),
|
||||
);
|
||||
reg.get_or_provision_shared(shared);
|
||||
|
||||
// Default tenant and a HELLO tenant both resolve to the shared store.
|
||||
let def = reg
|
||||
.get_or_provision(TenantId::from_str(TenantRegistry::DEFAULT_TENANT).unwrap())
|
||||
.unwrap();
|
||||
let hello = reg
|
||||
.get_or_provision(TenantId::from_str("alpha").unwrap())
|
||||
.unwrap();
|
||||
assert!(
|
||||
Arc::ptr_eq(&def.store, &hello.store),
|
||||
"legacy mode: all tenants share one store"
|
||||
);
|
||||
|
||||
// A write under HELLO is visible under the default (same store).
|
||||
let coord = cubecoords::Czyx::new(1, 2, 3, 4);
|
||||
hello
|
||||
.store
|
||||
.put_record(coord, &cubecoords::CubeHeader::new(), b"shared");
|
||||
assert_eq!(
|
||||
def.store.get_record(&coord).map(|(_, v)| v),
|
||||
Some(b"shared".to_vec())
|
||||
);
|
||||
hello.store.checkpoint();
|
||||
let _ = std::fs::remove_dir_all(&tmp);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user