nasaga_runtime_core/

registry.rs

//! 带不可变版本的 definition 注册表与启动预检。
//!
//! definition 一经发布内容不可变：同 `(workflow, version)` 注册不同摘要的
//! 内容会被拒绝。启动预检校验全部非终态实例引用的 `(workflow, version, digest)`
//! 可用且一致——缺失或漂移时拒绝 Ready，绝不用新内容驱动旧实例。

use std::collections::BTreeMap;

use nasaga_backend::SagaInstanceStore;
use nasaga_core::{
    DefinitionVersion, ResolutionMode, SagaId, TenantId, WorkflowDefinition, WorkflowName,
};

/// 业务作用：把不可变 definition 与“是否允许创建新实例”门禁绑定在同一快照中。
#[derive(Debug, Clone)]
struct DefinitionEntry {
    definition: WorkflowDefinition,
    accepts_new: bool,
}

/// 业务作用：持有本进程可驱动的全部 workflow definition，按 `(名称, 定义版本)` 索引。
///
/// 注册表只在启动阶段构建（`register` 全部完成后再交给 Orchestrator），运行期只读，
/// 因此无内部锁；定义演进并存表示同名 workflow 注册多个版本。
#[derive(Debug, Clone, Default)]
pub struct DefinitionRegistry {
    definitions: BTreeMap<(String, u32), DefinitionEntry>,
    tenant_definitions: BTreeMap<(String, String, u32), DefinitionEntry>,
}

impl DefinitionRegistry {
    /// 业务作用：创建空注册表。
    ///
    /// 参数说明: 无。
    ///
    /// 返回：不含任何 definition 的注册表。
    pub fn new() -> Self {
        Self::default()
    }

    /// 业务作用：判断当前快照是否没有任何流程定义，供动态 Catalog 空启动与静态模式门禁区分。
    ///
    /// 参数说明: 无。
    ///
    /// 返回：没有已注册定义时返回真。
    pub fn is_empty(&self) -> bool {
        self.definitions.is_empty() && self.tenant_definitions.is_empty()
    }

    /// 业务作用：读取当前不可变快照中的定义数量，供启动报告与 Catalog generation 核对。
    ///
    /// 参数说明: 无。
    ///
    /// 返回：按 `(workflow, definition_version)` 去重后的定义数量。
    pub fn len(&self) -> usize {
        self.definitions.len() + self.tenant_definitions.len()
    }

    /// 业务作用：按稳定键顺序遍历当前快照的全部流程定义，供发布器生成完整 definition seal。
    ///
    /// 参数说明: 无。
    ///
    /// 返回：以 workflow 名称、定义版本排序的定义迭代器。
    pub fn definitions(&self) -> impl Iterator<Item = &WorkflowDefinition> {
        self.definitions
            .values()
            .chain(self.tenant_definitions.values())
            .map(|entry| &entry.definition)
    }

    /// 业务作用：连同可选租户作用域遍历全部定义，供动态 capability 路由保持租户边界。
    ///
    /// 参数说明: 无。
    ///
    /// 返回：静态全局定义的租户为空，动态 Catalog 定义携带其规范租户。
    pub fn definitions_with_tenants(
        &self,
    ) -> impl Iterator<Item = (Option<&str>, &WorkflowDefinition)> {
        self.definitions
            .values()
            .map(|entry| (None, &entry.definition))
            .chain(
                self.tenant_definitions
                    .iter()
                    .map(|((tenant, _, _), entry)| (Some(tenant.as_str()), &entry.definition)),
            )
    }

    /// 业务作用：注册一个已通过构造预检的 definition，同一版本重复注册按摘要裁决。
    ///
    /// 参数说明：
    /// - `definition`: 待注册的流程定义。
    ///
    /// 返回：解决模式已具备 hosted 闭环且新增或摘要一致时返回 `Ok`；Callback/Manual
    /// 或同 `(workflow, version)` 摘要不一致返回错误，不允许带不完整合同启动。
    pub fn register(&mut self, definition: WorkflowDefinition) -> anyhow::Result<()> {
        validate_runtime_resolution_modes(&definition)?;
        let key = (
            definition.name().as_str().to_string(),
            definition.version().get(),
        );
        if let Some(existing) = self.definitions.get(&key) {
            // definition 不可变:同一版本出现两份不同内容说明发布纪律被破坏,
            // 静默用后者覆盖会让运行中实例被新内容驱动。
            if existing.definition.digest() != definition.digest() {
                anyhow::bail!(
                    "definition `{}` v{} already registered with a different digest",
                    key.0,
                    key.1
                );
            }
            return Ok(());
        }
        self.definitions.insert(
            key,
            DefinitionEntry {
                definition,
                accepts_new: true,
            },
        );
        Ok(())
    }

    /// 业务作用：为一个租户登记 active 或 deprecated definition，使新实例门禁与在途实例装载分离。
    ///
    /// 参数说明：
    /// - `tenant`：定义适用的租户身份。
    /// - `definition`：已经通过构造校验的不可变流程。
    /// - `accepts_new`：active 为真，deprecated 为假。
    ///
    /// 返回：首次登记或同键同摘要同状态时成功；任何原地内容或生命周期漂移返回错误。
    pub fn register_for_tenant(
        &mut self,
        tenant: TenantId,
        definition: WorkflowDefinition,
        accepts_new: bool,
    ) -> anyhow::Result<()> {
        validate_runtime_resolution_modes(&definition)?;
        let key = (
            tenant.as_str().to_owned(),
            definition.name().as_str().to_owned(),
            definition.version().get(),
        );
        if let Some(existing) = self.tenant_definitions.get(&key) {
            if existing.definition.digest() != definition.digest()
                || existing.accepts_new != accepts_new
            {
                anyhow::bail!(
                    "tenant definition `{}` v{} already registered with different content or lifecycle",
                    key.1,
                    key.2
                );
            }
            return Ok(());
        }
        self.tenant_definitions.insert(
            key,
            DefinitionEntry {
                definition,
                accepts_new,
            },
        );
        Ok(())
    }

    /// 业务作用：按名称与定义版本查找 definition，供推进裁决使用实例固定的内容。
    ///
    /// 参数说明：
    /// - `workflow`: workflow 名称。
    /// - `version`: definition 版本。
    ///
    /// 返回：存在返回定义引用；不存在返回 `None`。
    pub fn get(
        &self,
        workflow: &WorkflowName,
        version: DefinitionVersion,
    ) -> Option<&WorkflowDefinition> {
        self.definitions
            .get(&(workflow.as_str().to_string(), version.get()))
            .map(|entry| &entry.definition)
    }

    /// 业务作用：按租户优先读取运行期 definition，并为静态全租户定义保留兼容回退。
    ///
    /// 参数说明：
    /// - `tenant`：需要解析定义的租户身份。
    /// - `workflow`：实例绑定的工作流名。
    /// - `version`：实例绑定的业务定义版本。
    ///
    /// 返回：存在租户条目时返回其定义，不按 accepts_new 过滤；否则回退 get 的全局定义，均不存在时 None。只读注册表，不验证摘要。
    pub fn get_for_tenant(
        &self,
        tenant: &TenantId,
        workflow: &WorkflowName,
        version: DefinitionVersion,
    ) -> Option<&WorkflowDefinition> {
        self.tenant_definitions
            .get(&(
                tenant.as_str().to_owned(),
                workflow.as_str().to_owned(),
                version.get(),
            ))
            .map(|entry| &entry.definition)
            .or_else(|| self.get(workflow, version))
    }

    /// 业务作用：只返回当前允许创建新实例的租户 definition，deprecated 合同仍可供旧实例收敛。
    ///
    /// 参数说明：
    /// - `tenant`：需要解析定义的租户身份。
    /// - `workflow`：实例绑定的工作流名。
    /// - `version`：实例绑定的业务定义版本。
    ///
    /// 返回：存在租户条目时仅在 accepts_new 为 true 返回定义，拒绝时不回退全局；无租户条目才按全局 accepts_new 选择，未知或不接受新实例时 None。
    pub fn get_for_start(
        &self,
        tenant: &TenantId,
        workflow: &WorkflowName,
        version: DefinitionVersion,
    ) -> Option<&WorkflowDefinition> {
        if let Some(entry) = self.tenant_definitions.get(&(
            tenant.as_str().to_owned(),
            workflow.as_str().to_owned(),
            version.get(),
        )) {
            return entry.accepts_new.then_some(&entry.definition);
        }
        self.definitions
            .get(&(workflow.as_str().to_owned(), version.get()))
            .filter(|entry| entry.accepts_new)
            .map(|entry| &entry.definition)
    }

    /// 业务作用：启动预检——校验全部非终态实例引用的 definition 可用且摘要一致。 缺失或摘要漂移时必须拒绝 Ready 而不是带病启动：用新内容驱动旧实例会让补偿 顺序、身份派生与冻结计划全部失真。
    ///
    /// 参数说明：
    /// - `store`：用于分页读取非终态实例的持久化后端。
    /// - `scan_limit`：单次扫描的实例条数上限，须由调用方确保非零且后端支持。
    ///
    /// 返回：本次分页扫描遇到的实例定义都存在且摘要一致时成功；缺失或漂移返回聚合计数，读取失败原样传播 store 错误。不建立跨页一致快照，scan_limit 本层不校验非零。
    pub async fn verify_non_terminal<S>(&self, store: &S, scan_limit: u32) -> anyhow::Result<()>
    where
        S: SagaInstanceStore,
    {
        let mut missing = 0usize;
        let mut drifted = 0usize;
        let mut cursor: Option<SagaId> = None;
        loop {
            let instances = store
                .list_non_terminal_after(cursor.as_ref(), scan_limit)
                .await?;
            if instances.is_empty() {
                break;
            }
            for instance in &instances {
                match self.get_for_tenant(
                    &instance.tenant,
                    &instance.workflow,
                    instance.definition_version,
                ) {
                    None => missing += 1,
                    Some(definition) if definition.digest() != instance.definition_digest => {
                        drifted += 1
                    }
                    Some(_) => {}
                }
            }
            // 以本页末尾身份续读，避免 OFFSET 因前页数量变化而位移；这不构成跨页快照，
            // 并发出现且位于游标之前的实例仍需宿主通过启动期写入边界或后续复验约束。
            cursor = instances.last().map(|instance| instance.saga_id.clone());
            if instances.len() < scan_limit as usize {
                break;
            }
        }
        if missing > 0 || drifted > 0 {
            anyhow::bail!(
                "definition precheck failed: {missing} non-terminal instance(s) reference \
                 unregistered definitions, {drifted} reference drifted digests; refusing Ready"
            );
        }
        Ok(())
    }
}

/// 业务作用：拒绝当前 Orchestrator 尚无可靠接入闭环的 Callback/Manual 解决模式。
///
/// runtime 目前只会为 `Poll` 原子登记 attempt、发布 resolve Outbox 并布置 timeout；若接收
/// Callback/Manual definition，结果没有已登记 command 锚点，实例只能在预算耗尽后转人工。
/// 与其运行时悬挂，必须在注册阶段 fail-fast。
///
/// 参数说明：
/// - `definition`: 待加入运行时注册表的已校验 definition。
///
/// 返回：所有允许 Unknown 的步骤都使用 `Poll` 时返回 `Ok`；否则返回启动配置错误。
fn validate_runtime_resolution_modes(definition: &WorkflowDefinition) -> anyhow::Result<()> {
    if let Some(step) = definition.steps().iter().find(|step| {
        step.resolution().allow_unknown() && step.resolution().mode() != Some(ResolutionMode::Poll)
    }) {
        anyhow::bail!(
            "definition `{}` v{} step `{}` uses an unsupported Saga runtime resolution mode; only poll is hosted",
            definition.name().as_str(),
            definition.version().get(),
            step.name().as_str()
        );
    }
    Ok(())
}