Method submit

nodedb-wal/src/group_commit.rs:106–188

Submit a write and block until it's durable. Returns the assigned LSN once the batch containing this write has been fsynced to disk. If fsync fails, the error is propagated to all threads whose writes were in the failed batch.

(&self, writer: &Mutex<WalWriter>, write: PendingWrite)

    /// fsynced to disk. If fsync fails, the error is propagated to all threads
    /// whose writes were in the failed batch.
    pub fn submit(&self, writer: &Mutex<WalWriter>, write: PendingWrite) -> Result<CommitResult> {
        // Enqueue the write.
        {
            let mut pending = self.pending.lock().map_err(|_| WalError::LockPoisoned {
                context: "pending queue",
            })?;
            pending.push(write);
        }

        // Try to become the commit leader. If another thread holds the commit
        // lock, we block here until it finishes — then we check if our write
        // was already committed by that leader.
        let _commit_guard = self
            .commit_lock
            .lock()
            .map_err(|_| WalError::LockPoisoned {
                context: "commit lock",
            })?;

        // Drain pending writes. If the previous leader already committed our
        // write, the pending queue will be empty (our write was drained by
        // that leader). If the queue is non-empty, we are the new leader.
        let batch: Vec<PendingWrite> = {
            let mut pending = self.pending.lock().map_err(|_| WalError::LockPoisoned {
                context: "pending queue (drain)",
            })?;
            std::mem::take(&mut *pending)
        };

        if batch.is_empty() {
            // Previous leader drained our write. Check if that commit
            // succeeded or failed. This is the critical fix: without this
            // check, a non-leader would return durable:true even if the
            // leader's fsync failed.
            if self.last_commit_failed.load(Ordering::Acquire) {
                return Err(WalError::Io(std::io::Error::other(
                    "WAL fsync failed in previous group commit batch",
                )));
            }
            let lsn = self.durable_lsn.load(Ordering::Acquire);
            return Ok(CommitResult { lsn, durable: true });
        }

        // We are the leader — append and fsync.
        let mut wal = writer.lock().map_err(|_| WalError::LockPoisoned {
            context: "WAL writer",
        })?;
        let mut last_lsn = 0;

        for w in &batch {
            last_lsn = wal.append(
                w.record_type,
                w.tenant_id,
                w.vshard_id,
                w.database_id,
                &w.payload,
            )?;
        }