Function wal_append_timeseries

nodedb/src/control/server/wal_dispatch.rs:530–559

Append a timeseries batch to WAL and return the assigned LSN. Used by the ILP listener to propagate the WAL LSN to the Data Plane for proper dedup tracking and `flush_wal_lsn` in partition metadata. Returns `None` if WAL is bypassed for this collection.

(
    wal: &WalManager,
    tenant_id: TenantId,
    vshard_id: VShardId,
    database_id: DatabaseId,
    collection: &str,
    payload: &[u8],
    credentials: Option<&CredentialStore>,
)

/// for proper dedup tracking and `flush_wal_lsn` in partition metadata.
/// Returns `None` if WAL is bypassed for this collection.
pub fn wal_append_timeseries(
    wal: &WalManager,
    tenant_id: TenantId,
    vshard_id: VShardId,
    database_id: DatabaseId,
    collection: &str,
    payload: &[u8],
    credentials: Option<&CredentialStore>,
) -> crate::Result<Option<nodedb_types::Lsn>> {
    // WAL bypass check.
    if let Some(creds) = credentials
        && let Some(catalog) = creds.catalog()
        && let Ok(Some(coll)) = catalog.get_collection(database_id, tenant_id.as_u64(), collection)
        && let Some(config) = coll.get_timeseries_config()
        && config.get("wal").and_then(|v| v.as_str()) == Some("false")
    {
        return Ok(None);
    }

    let payload_vec = payload.to_vec();
    let wal_payload =
        zerompk::to_msgpack_vec(&("timeseries", collection, payload_vec)).map_err(|e| {
            crate::Error::Serialization {
                format: "msgpack".into(),
                detail: format!("wal timeseries batch: {e}"),
            }
        })?;
    let lsn = wal.append_timeseries_batch(tenant_id, vshard_id, database_id, &wal_payload)?;
    Ok(Some(lsn))
}