Method atomic_put

nodedb/src/engine/kv/engine_atomic.rs:319–399

(
        &mut self,
        tenant_id: u64,
        collection: &str,
        tkey: u64,
        key: &[u8],
        value: &[u8],
        ttl_ms: u64,
        now_ms: u64,
        is_new_key: bool,

    /// If `ttl_ms > 0`, sets/resets TTL. If key is new and `ttl_ms == 0`, no TTL.
    #[allow(clippy::too_many_arguments)]
    fn atomic_put(
        &mut self,
        tenant_id: u64,
        collection: &str,
        tkey: u64,
        key: &[u8],
        value: &[u8],
        ttl_ms: u64,
        now_ms: u64,
        is_new_key: bool,
    ) {
        // Cache metadata lookup to avoid double HashMap access.
        let old_meta = if is_new_key {
            None
        } else {
            self.tables.get(&tkey).and_then(|t| t.get_entry_meta(key))
        };

        // Determine the target expire_at.
        let expire_at = if ttl_ms > 0 {
            // Explicit TTL: set/reset.
            now_ms + ttl_ms
        } else if let Some(ref meta) = old_meta {
            // Existing key, preserve TTL.
            meta.expire_at_ms
        } else {
            // New key with no TTL request: persistent.
            NO_EXPIRY
        };

        // Cancel old expiry before mutation.
        if let Some(ref meta) = old_meta
            && meta.has_ttl
        {
            let composite = expiry_key(tenant_id, collection, key);
            self.expiry.cancel(&composite, meta.expire_at_ms);
        }

        // Extract old field values BEFORE overwriting — needed so on_put can
        // remove stale index entries when a field changes.
        let old_fields =
            if !is_new_key && self.indexes.get(&tkey).is_some_and(|idx| !idx.is_empty()) {
                self.tables
                    .get(&tkey)
                    .and_then(|t| t.get(key, now_ms))
                    .map(|old_val| {
                        super::engine_helpers::extract_all_field_values_from_msgpack(old_val)
                    })
            } else {
                None
            };

        // Write the value.
        let table = self.tables.get_mut(&tkey).expect("table ensured");
        table.put(key, value, expire_at, nodedb_types::Surrogate::ZERO);

        // Schedule new expiry if needed.
        if expire_at != NO_EXPIRY {