Function apply_op

nodedb-array/src/sync/apply.rs:119–172

Apply `op` to `engine`, returning an outcome without panicking. Steps: 1. Shape validation — on error: `Rejected(ShapeInvalid)`. 2. Schema HLC check — `None` → `Rejected(ArrayUnknown)`; `op.header.schema_hlc > local` → `Rejected(SchemaTooNew)`. 3. Idempotency — already seen → `Idempotent`. 4. Dispatch to `apply_put`/`apply_delete`/`apply_erase`. Engine errors that indicate corruption (`SegmentCor

(engine: &mut E, op: &ArrayOp)

/// 5. Tile-cache invalidation.
/// 6. Return `Applied`.
pub fn apply_op<E: ApplyEngine>(engine: &mut E, op: &ArrayOp) -> ArrayResult<ApplyOutcome> {
    // 1. Shape validation.
    if let Err(e) = op.validate_shape() {
        return Ok(ApplyOutcome::Rejected(ApplyRejection::ShapeInvalid {
            detail: e.to_string(),
        }));
    }

    // 2. Schema HLC gating.
    match engine.schema_hlc(&op.header.array)? {
        None => {
            return Ok(ApplyOutcome::Rejected(ApplyRejection::ArrayUnknown {
                name: op.header.array.clone(),
            }));
        }
        Some(local_schema) if op.header.schema_hlc > local_schema => {
            return Ok(ApplyOutcome::Rejected(ApplyRejection::SchemaTooNew {
                local: local_schema,
                op: op.header.schema_hlc,
            }));
        }
        Some(_) => {}
    }

    // 3. Idempotency.
    if engine.already_seen(&op.header.array, op.header.hlc)? {
        return Ok(ApplyOutcome::Idempotent);
    }

    // 4. Dispatch.
    let dispatch_result = match op.kind {
        ArrayOpKind::Put => engine.apply_put(op),
        ArrayOpKind::Delete => engine.apply_delete(op),
        ArrayOpKind::Erase => engine.apply_erase(op),
    };

    if let Err(e) = dispatch_result {
        // Corruption-grade errors propagate; everything else becomes a rejection.
        match &e {
            ArrayError::SegmentCorruption { .. } | ArrayError::HlcLockPoisoned => return Err(e),
            _ => {
                return Ok(ApplyOutcome::Rejected(ApplyRejection::EngineRejected {
                    detail: e.to_string(),
                }));
            }
        }
    }

    // 5. Tile invalidation.
    engine.invalidate_tile(&op.header.array, &op.coord)?;

    // 6. Success.
    Ok(ApplyOutcome::Applied)
}

// ─── MockEngine (test / test-utils only) ────────────────────────────────────