Method assign

nodedb/src/control/surrogate/assign.rs:99–152

Resolve `(collection, pk_bytes)` to a stable surrogate. - If the credential store has no catalog (in-memory test fixture), returns `Surrogate::ZERO`. Production state always wires a redb-backed `CredentialStore::open` so this branch never fires. - If a binding already exists, return it (no allocation, no flush). - Else: allocate one surrogate, persist the binding, and check the registry's flush t

(&self, collection: &str, pk_bytes: &[u8])

    /// on the registry write-lock so the registry hwm and the
    /// persisted PK row cannot diverge under concurrent assigners.
    pub fn assign(&self, collection: &str, pk_bytes: &[u8]) -> crate::Result<Surrogate> {
        let catalog = match self.credential_store.catalog().as_ref() {
            Some(c) => c,
            None => return Ok(Surrogate::ZERO),
        };

        // Fast-path: existing binding. Done under a read lock — most
        // production calls land here once the per-collection working
        // set has been observed.
        if let Some(s) = catalog.get_surrogate_for_pk(DatabaseId::DEFAULT, collection, pk_bytes)? {
            return Ok(s);
        }

        // Slow path: allocate + persist + maybe flush. The write lock
        // guards the (allocate, write-pk-row) pair so two concurrent
        // assigners can't both observe "missing", both allocate, and
        // both write — the second would silently overwrite the
        // first's binding with a different surrogate.
        let registry = self.registry.write().map_err(|_| crate::Error::Internal {
            detail: "surrogate registry lock poisoned".into(),
        })?;
        // Re-check inside the lock: another assigner may have raced
        // us between the read above and the lock acquisition.
        if let Some(s) = catalog.get_surrogate_for_pk(DatabaseId::DEFAULT, collection, pk_bytes)? {
            return Ok(s);
        }
        let surrogate = registry.alloc_one()?;
        catalog.put_surrogate(DatabaseId::DEFAULT, collection, pk_bytes, surrogate)?;
        // Emit a durable WAL bind before the lock releases. Order is
        // load-bearing: a crash between catalog write and bind append
        // is invisible (the catalog row is already on disk via redb's
        // own WAL); a crash before the catalog write leaves nothing
        // to recover; a crash between bind append and lock release is
        // recovered by replaying the bind into the catalog (idempotent
        // via the two-table overwrite).
        self.wal_appender
            .record_bind_to_wal(surrogate.as_u32(), collection, pk_bytes)?;

        // Flush trigger: durably checkpoint the new hwm if either the
        // ops or elapsed-time threshold has tripped. Both writes are
        // idempotent so a crash between them on a re-run replays
        // cleanly.
        if registry.should_flush() {
            let raft_shared = self.shared.get().and_then(|w| w.upgrade());
            let combined = CombinedPersist {
                catalog,
                wal_appender: self.wal_appender.as_ref(),
                raft_shared: raft_shared.as_deref(),
            };
            registry.flush(&combined)?;
        }

        Ok(surrogate)
    }

    /// Read-only lookup: return the surrogate previously bound to
    /// `(collection, pk_bytes)` without ever allocating or writing.