Struct Writer 

pub struct Writer {
    pub l0_manager: Arc<L0Manager>,
    pub storage: Arc<StorageManager>,
    pub schema_manager: Arc<SchemaManager>,
    pub allocator: Arc<IdAllocator>,
    pub config: UniConfig,
    pub property_manager: Option<Arc<PropertyManager>>,
    pub fork_id: Option<ForkId>,
    /* private fields */
}

Fields

l0_manager: Arc<L0Manager>

storage: Arc<StorageManager>

schema_manager: Arc<SchemaManager>

allocator: Arc<IdAllocator>

config: UniConfig

property_manager: Option<Arc<PropertyManager>>

Property manager for cache invalidation after flush

fork_id: Option<ForkId>

Identifier of the fork this writer serves, if any. None for primary’s writer. Set by crate::fork::writer_factory::new_for_fork and read in flush_to_l1 to emit fork-tagged metrics and to fire the fragment-count guard rail (Phase 2 Day 12).

Implementations

impl Writer

pub async fn new( storage: Arc<StorageManager>, schema_manager: Arc<SchemaManager>, start_version: u64, ) -> Result<Self>

pub async fn new_with_config( storage: Arc<StorageManager>, schema_manager: Arc<SchemaManager>, start_version: u64, config: UniConfig, wal: Option<Arc<WriteAheadLog>>, allocator: Option<Arc<IdAllocator>>, ) -> Result<Self>

pub fn row_lock_handle(&self, key: &[u8]) -> Arc<Mutex<()>>

Returns the shared pessimistic lock handle for a FOR UPDATE row key, creating it on first use. The caller .lock_owned().awaits the returned mutex and holds the guard for the transaction’s lifetime.

pub fn release_for_update_locks(&self, keys: &[Vec<u8>])

Prunes FOR UPDATE lock-map entries for keys that no live transaction holds anymore, so the map does not grow without bound across the keyspace.

Called when a transaction ends, after its guards have been dropped. remove_if evaluates its predicate under the DashMap shard lock, which is the same lock row_lock_handle takes to clone an entry — so the check strong_count == 1 (only the map holds the Arc) is race-free: a concurrent acquirer either already cloned the Arc (count ≥ 2 → we skip removal) or has not yet taken the shard lock (it will mint a fresh entry after we remove). Either way no two transactions ever lock different Mutex instances for the same key.

pub fn for_update_lock_count(&self) -> usize

Number of live entries in the FOR UPDATE lock map. Introspection for tests that the map does not leak entries across transactions (G5).

pub fn current_commit_sequence(&self) -> u64

The current OCC commit sequence. A FOR UPDATE acquisition re-stamps a fresh transaction’s occ_read_seq to this so its conflict-detection baseline advances to lock-acquisition time (read-latest under the lock).

pub fn flush_coordinator(&self) -> Option<&Arc<FlushCoordinator>>

Borrow the flush coordinator if async flush is enabled. Returns None when config.async_flush_enabled = false. External callers (drop_fork) use this to drain pending streams.

pub fn set_index_rebuild_manager( &self, manager: Arc<IndexRebuildManager>, ) -> Result<()>

Set the index rebuild manager for post-flush automatic rebuild scheduling.

One-shot: returns Err if already set. The receiver is &self so this can be called after the Writer has been wrapped in Arc<Writer>.

pub async fn replay_wal(&self, wal_high_water_mark: u64) -> Result<usize>

Replay WAL mutations into the current L0 buffer.

pub async fn next_vid(&self) -> Result<Vid>

Allocates the next VID (pure auto-increment).

pub async fn allocate_vids(&self, count: usize) -> Result<Vec<Vid>>

Allocates multiple VIDs at once for bulk operations. This is more efficient than calling next_vid() in a loop.

pub async fn next_eid(&self, _type_id: u32) -> Result<Eid>

Allocates the next EID (pure auto-increment).

pub async fn allocate_eids(&self, count: usize) -> Result<Vec<Eid>>

Allocates multiple EIDs at once for bulk operations. This is more efficient than calling next_eid() in a loop.

pub fn set_xervo_runtime(&self, runtime: Arc<ModelRuntime>) -> Result<()>

Install the embedding runtime exactly once. Receiver is &self so it can be called after the Writer has been wrapped in Arc<Writer>.

pub fn xervo_runtime(&self) -> Option<Arc<ModelRuntime>>

pub fn create_transaction_l0(&self) -> Arc<RwLock<L0Buffer>>

Create a new empty L0 buffer for transaction-scoped mutations.

Only reads the current version — no exclusive lock required on Writer. The returned buffer has no WAL reference; mutations are logged at commit time via Self::commit_transaction_l0.

pub async fn commit_transaction_l0( self: &Arc<Self>, tx_l0_arc: Arc<RwLock<L0Buffer>>, ) -> Result<(u64, bool)>

Commit an externally-owned transaction L0 buffer.

Writes mutations to WAL, flushes, merges into main L0, and replays edges into the AdjacencyManager. Returns the WAL LSN of the commit (0 when no WAL is configured). Commit a transaction’s private L0 buffer into main L0.

Returns (wal_lsn, flush_pending). When flush_pending == true, the post-commit should_flush() predicate fired but no flush ran — the caller is expected to spawn a background flush_to_l1. This is the shape used when UniConfig::async_flush_enabled is set, so commits don’t block on L1-streaming I/O.

pub async fn flush_wal(&self) -> Result<u64>

Flush the WAL buffer to durable storage.

Returns the LSN of the flushed segment, or 0 when no WAL is configured.

pub fn track_properties_removed( &self, count: usize, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, )

Record property removals in the active L0 mutation stats.

Routes to the transaction L0 if provided, otherwise to the main L0.

pub async fn get_vertex_labels( &self, vid: Vid, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Option<Vec<String>>

Get vertex labels from all sources: current L0, pending L0s, and storage. This is the proper way to read vertex labels after a flush, as it checks both in-memory buffers and persisted storage.

pub fn get_edge_type_id_from_l0( &self, eid: Eid, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Option<u32>

Look up the edge type ID (u32) for an EID from the L0 buffer’s edge endpoints. Falls back to the transaction L0 if available.

pub fn set_edge_type( &self, eid: Eid, type_name: String, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, )

Set the type name for an edge (used for schemaless edge types). This is called during CREATE for edge types not found in the schema.

pub async fn insert_vertex( &self, vid: Vid, properties: Properties, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

pub async fn insert_vertex_with_labels( &self, vid: Vid, properties: Properties, labels: &[String], tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<Properties>

pub async fn insert_vertex_partial_full( &self, vid: Vid, props: Properties, touched_keys: HashSet<String>, labels: &[String], tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

Insert a vertex’s FULL property row plus a touched-keys hint so the flush emits ONLY those columns via Lance MergeInsert.

Caller must have read the full row (via PropertyManager) and applied SET-touched values on top before calling — same input shape as insert_vertex_with_labels. The new arg touched_keys is the set of property keys this SET statement actually assigned; L0 records it in vertex_partial_keys[vid] and the flush filters the MergeInsert source schema down to those keys. When UniConfig::partial_lance_writes == false, falls through to insert_vertex_with_labels (Append) — preserving bit-for-bit equivalence with prior releases.

pub async fn insert_vertex_partial( &self, vid: Vid, touched: Properties, labels: &[String], tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

Insert a vertex’s partial property set without first reading the full row.

When WriterConfig::partial_lance_writes is true, the touched keys flow into L0Buffer::vertex_partial_keys so the next flush emits them via Lance MergeInsertBuilder against a subset-of- schema source — preserving untouched columns (e.g., embeddings) byte-equal in Lance with no read at the caller and no write of those columns.

When the flag is false, this falls back to the existing insert_vertex_with_labels path after merging touched with the current properties from L0/storage. The caller can therefore use this entry point unconditionally; the optimization activates only when the flag is on.

pub async fn insert_vertices_batch( &self, vids: Vec<Vid>, properties_batch: Vec<Properties>, labels: Vec<String>, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<Vec<Properties>>

Insert multiple vertices with batched operations.

This method uses batched operations to achieve O(N) complexity instead of O(N²) for bulk inserts with unique constraints.

Performance Improvements

• Batch VID allocation: 1 call instead of N calls
• Batch constraint validation: O(N) instead of O(N²)
• Batch embedding generation: 1 API call per config instead of N calls
• Transaction wrapping: Automatic flush deferral, atomicity

Arguments

• vids - Pre-allocated VIDs for the vertices
• properties_batch - Properties for each vertex
• labels - Labels for all vertices (assumes single label for simplicity)

Errors

Returns error if:

• VID/properties length mismatch
• Constraint violation detected
• Embedding generation fails
• Transaction commit fails

Atomicity

If this method fails, all changes are rolled back (if transaction was started here).

pub async fn delete_vertex( &self, vid: Vid, labels: Option<Vec<String>>, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

Delete a vertex by VID.

When labels is provided, uses them directly to populate L0 for correct tombstone flushing. Otherwise discovers labels from L0 buffers and storage (which can be slow for many vertices).

Errors

Returns an error if write pressure stalls, label lookup fails, or the L0 delete operation fails.

pub async fn insert_edge_partial_full( &self, src_vid: Vid, dst_vid: Vid, edge_type: u32, eid: Eid, props: Properties, edge_type_name: Option<String>, touched_keys: HashSet<String>, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

pub async fn insert_edge( &self, src_vid: Vid, dst_vid: Vid, edge_type: u32, eid: Eid, properties: Properties, edge_type_name: Option<String>, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

pub async fn delete_edge( &self, eid: Eid, src_vid: Vid, dst_vid: Vid, edge_type: u32, tx_l0: Option<&Arc<RwLock<L0Buffer>>>, ) -> Result<()>

pub async fn check_flush(&self) -> Result<()>

Check if flush should be triggered based on mutation count or time elapsed. This method is called after each write operation and can also be called by a background task for time-based flushing.

pub async fn flush_to_l1(&self, name: Option<String>) -> Result<String>

Flushes the current in-memory L0 buffer to L1 storage.

Lock Ordering

To prevent deadlocks, locks must be acquired in the following order:

1. Writer lock (held by caller via outer Arc<RwLock<Writer>>; removed in Phase 4)
2. flush_lock (acquired by this entry point; held across the whole flush)
3. L0Manager lock (via begin_flush / get_current)
4. L0Buffer lock (individual buffer RWLocks)
5. Index / Storage locks (during actual flush)

Callers that already hold flush_lock (today only commit_transaction_l0) must call flush_inline_under_lock (private) directly to avoid a re-entrant tokio::sync::Mutex deadlock — see concurrent_writer.md §5.5.

pub async fn flush_to_l1_async( self: &Arc<Self>, name: Option<String>, ) -> Result<FlushTicket>

Async-flush entry point: rotate under flush_lock, release the lock, then submit the stream phase to the FlushCoordinator. Returns a FlushTicket that resolves when finalize completes.

Errors if config.async_flush_enabled = false (the coordinator is None in that case — see flush_coordinator field doc).