Skip to main content

nadisc_runtime/
registration.rs

1//! 注册写入串行化与本地资格。远端结果不明时停止相反写入后的重新发布。
2
3use crate::{
4    changed, closed, config::RegistrationPolicy, lock, millis, operation, timeout, Connection,
5    Core, Gate, OwnerPhase,
6};
7use nadisc::managed::{
8    DiscoveryError, ErrorCode, RegistrationEvidence, RegistrationPlan, Result, WriteAdmission,
9};
10use std::{
11    sync::{
12        atomic::{AtomicBool, AtomicU64, Ordering},
13        Arc, Mutex,
14    },
15    time::Duration,
16};
17use tokio::{sync::watch, time::Instant};
18
19/// 宿主提供的本地业务资格,不应由包含远端注册结果的整体 readiness 反向驱动。
20#[derive(Clone)]
21pub struct Eligibility {
22    inner: Arc<Mutex<bool>>,
23    changes: watch::Sender<bool>,
24    check: Option<Arc<dyn Fn() -> bool + Send + Sync>>,
25    valid_until: Arc<Mutex<Option<Instant>>>,
26    source: Option<std::sync::Weak<()>>,
27}
28
29/// 手动宿主的唯一资格发送者;丢弃后读句柄立即失效。
30pub struct EligibilityController {
31    signal: Eligibility,
32    _lifetime: Arc<()>,
33}
34impl EligibilityController {
35    /// 业务作用:发布有期限的本地资格,不让接收者保留永久有效的最后成功值。
36    ///
37    /// 参数说明:
38    /// - `ready`:为本地事实
39    /// - `valid_for`:必须在一毫秒到三十秒之间。
40    ///
41    /// 返回:有效输入同步改变资格;非法期限拒绝且不开放接流。
42    pub fn update(&self, ready: bool, valid_for: Duration) -> Result<()> {
43        if valid_for < Duration::from_millis(1) || valid_for > Duration::from_secs(30) {
44            self.signal.set(false);
45            return Err(DiscoveryError::new(
46                ErrorCode::InvalidConfig,
47                "eligibility validity is out of bounds",
48            ));
49        }
50        let mut value = lock(&self.signal.inner);
51        *value = ready;
52        *lock(&self.signal.valid_until) = ready.then(|| Instant::now() + valid_for);
53        self.signal.changes.send_replace(ready);
54        Ok(())
55    }
56}
57impl Drop for EligibilityController {
58    /// 业务作用:发送者退出时立即撤销发布许可并唤醒注册监督。
59    ///
60    /// 参数说明:无。
61    ///
62    /// 返回:保留的读句柄不能延续最后一次 true。
63    fn drop(&mut self) {
64        self.signal.set(false);
65    }
66}
67
68impl Eligibility {
69    /// 业务作用:创建默认不接流的资格源,监听器准备后再显式开放。
70    ///
71    /// 参数说明:无。
72    ///
73    /// 返回:可由宿主保留的资格控制句柄。
74    pub fn new() -> Self {
75        Self {
76            inner: Arc::new(Mutex::new(false)),
77            changes: watch::channel(false).0,
78            check: None,
79            valid_until: Arc::new(Mutex::new(None)),
80            source: None,
81        }
82    }
83
84    /// 业务作用:为手动宿主分离资格更新责任与传入 owner 的只读信号。
85    ///
86    /// 参数说明:无。
87    ///
88    /// 返回:唯一发送者和默认不接流的信号,发送者退出或证据过期均撤权。
89    pub fn channel() -> (EligibilityController, Self) {
90        let lifetime = Arc::new(());
91        let mut signal = Self::new();
92        signal.source = Some(Arc::downgrade(&lifetime));
93        (
94            EligibilityController {
95                signal: signal.clone(),
96                _lifetime: lifetime,
97            },
98            signal,
99        )
100    }
101
102    /// 业务作用:同步改变本地发布资格,与注册提交前复验使用同一把锁。
103    ///
104    /// 参数说明:
105    /// - `ready`:表示该注册对应的本地 listener 和业务依赖是否可接流。
106    ///
107    /// 返回:后续发布立即受约束;手动 true 最多有效三十秒,需要宿主持续更新。
108    pub fn set(&self, ready: bool) {
109        let mut value = lock(&self.inner);
110        *value = ready;
111        *lock(&self.valid_until) = ready.then(|| Instant::now() + Duration::from_secs(30));
112        self.changes.send_replace(ready);
113    }
114
115    /// 业务作用:读取本地资格供独立健康端点返回真实状态。
116    ///
117    /// 参数说明:无。
118    ///
119    /// 返回:本地事实,不包含注册中心回查结果。
120    pub fn is_ready(&self) -> bool {
121        self.current(*lock(&self.inner))
122    }
123
124    /// 业务作用:绑定宿主可同步读取的本地资格,在实际发布点重新读取而不依赖轮询延迟。
125    ///
126    /// 参数说明:
127    /// - `check`:必须无网络、有限时间返回,且不得重入 discovery 或改变资格对象。
128    ///
129    /// 返回:只依赖宿主当前事实的资格源;闭包必须自行包含停机和证据期限门禁。
130    pub fn from_check(check: impl Fn() -> bool + Send + Sync + 'static) -> Self {
131        Self {
132            inner: Arc::new(Mutex::new(true)),
133            changes: watch::channel(true).0,
134            check: Some(Arc::new(check)),
135            valid_until: Arc::new(Mutex::new(None)),
136            source: None,
137        }
138    }
139
140    /// 业务作用:在已持有本地许可锁时复验证据期限或宿主实时状态。
141    ///
142    /// 参数说明:
143    /// - `ready`:是保护范围内的显式许可。
144    ///
145    /// 返回:停止或证据到期均为 false。
146    fn current(&self, ready: bool) -> bool {
147        ready
148            && self
149                .source
150                .as_ref()
151                .is_none_or(|source| source.strong_count() > 0)
152            && self.check.as_ref().map_or_else(
153                || lock(&self.valid_until).is_some_and(|until| Instant::now() < until),
154                |check| check(),
155            )
156    }
157}
158
159impl Default for Eligibility {
160    /// 业务作用:默认关闭发布许可,避免构造组件即对外注册。
161    ///
162    /// 参数说明:无。
163    ///
164    /// 返回:未就绪的本地资格源。
165    fn default() -> Self {
166        Self::new()
167    }
168}
169
170/// 冻结前的注册计划、目标连接与本地发布资格。
171pub struct RegistrationSpec {
172    /// 目标连接的 qualifier,必须引用 owner 中已配置的连接。
173    pub connection_ref: String,
174    /// 冻结后提交给后端的完整注册计划。
175    pub plan: RegistrationPlan,
176    /// 目标连接或注册的冻结策略。
177    pub policy: RegistrationPolicy,
178    /// 发布前复验的本地业务资格,不依赖远端注册健康回查。
179    pub eligibility: Eligibility,
180}
181
182/// 单个注册目标的发布、健康确认与注销状态。
183#[derive(Clone, Copy, Debug, Eq, PartialEq)]
184pub enum RegistrationState {
185    /// 计划已冻结,尚未写入后端。
186    Planned,
187    /// 正在提交注册写入。
188    Registering,
189    /// 远端写入结果不明,需要协调后才能继续发布。
190    OutcomeUnknown,
191    /// 已提交注册,尚未获得有效健康证据。
192    PendingHealth,
193    /// 已获得健康证据,仍需按 valid_until 判断时效。
194    Active,
195    /// 本地资格或后端证据不再允许继续接流。
196    Suspended,
197    /// 已停止发布,正在注销远端实例。
198    Withdrawing,
199    /// 注销未能确认完成,可能存在远端残留。
200    Residual,
201    /// 已确认目标注销完成。
202    Removed,
203}
204
205/// 注册目标的只读证据摘要;Active 仍需检查证据有效期。
206#[derive(Clone, Debug)]
207pub struct RegistrationStatus {
208    /// 注册目标的本地唯一标识。
209    pub id: String,
210    /// 承载该注册的连接 qualifier。
211    pub connection: String,
212    /// 当前注册阶段;远端写入结果不明时不能当作成功。
213    pub state: RegistrationState,
214    /// 是否曾完成启动所要求的首次注册确认。
215    pub startup_confirmed: bool,
216    /// 远端注册失去有效健康证据时是否影响宿主整体就绪。
217    pub readiness_critical: bool,
218    /// 最近一次有效健康证据的本地单调时刻。
219    pub observed_at: Option<Instant>,
220    /// 当前健康证据的截止时刻;缺失或过期均不代表就绪。
221    pub valid_until: Option<Instant>,
222    /// 最近的注册或回查错误,不包含凭据原文。
223    pub error: Option<DiscoveryError>,
224}
225
226impl RegistrationStatus {
227    /// 业务作用:读取无需网络访问的当前发布证据。
228    ///
229    /// 参数说明:无。
230    ///
231    /// 返回:已确认为 Active 且证据尚未到期才算远端就绪。
232    pub fn is_ready(&self) -> bool {
233        self.state == RegistrationState::Active
234            && self
235                .valid_until
236                .is_some_and(|deadline| Instant::now() < deadline)
237    }
238}
239
240pub(crate) struct Record {
241    pub spec: RegistrationSpec,
242    pub connection: Arc<Connection>,
243    pub status: Mutex<RegistrationStatus>,
244    pub started: AtomicBool,
245    pub confirmed: AtomicBool,
246    health_revocation: AtomicU64,
247    attempted: AtomicBool,
248    uncertain: AtomicBool,
249    unsettled_renewal: AtomicBool,
250}
251
252impl Record {
253    /// 业务作用:在第一项外部写入前持有完整计划和状态记录。
254    ///
255    /// 参数说明:
256    /// - `spec`:已完成预检
257    /// - `connection`:提供真实后端身份。
258    ///
259    /// 返回:尚未发布的注册记录。
260    pub(crate) fn new(spec: RegistrationSpec, connection: Arc<Connection>) -> Self {
261        let health_revocation = connection.revocation.load(Ordering::Acquire);
262        Self {
263            status: Mutex::new(RegistrationStatus {
264                id: spec.plan.id.clone(),
265                connection: spec.connection_ref.clone(),
266                state: RegistrationState::Planned,
267                startup_confirmed: false,
268                readiness_critical: spec.policy.readiness_critical,
269                observed_at: None,
270                valid_until: None,
271                error: None,
272            }),
273            spec,
274            connection,
275            started: AtomicBool::new(false),
276            confirmed: AtomicBool::new(false),
277            health_revocation: AtomicU64::new(health_revocation),
278            attempted: AtomicBool::new(false),
279            uncertain: AtomicBool::new(false),
280            unsettled_renewal: AtomicBool::new(false),
281        }
282    }
283
284    /// 业务作用:统一裁决启动确认、宿主就绪与诊断读取的注册健康资格。
285    ///
286    /// 参数说明:
287    /// - `gate`:是 owner 已持有的状态门禁,与连接撤权及健康证据发布串行。
288    ///
289    /// 返回:旧身份代次、本地失权或停机的证据不能继续用于就绪;保留注销残留等独立生命周期事实。
290    pub(crate) fn current_status(&self, gate: &Gate) -> RegistrationStatus {
291        let mut status = lock(&self.status).clone();
292        // 身份代次只由有效的新回查推进;复制状态不能使被撤销的旧证据重新有效。
293        if self.health_revocation.load(Ordering::Acquire)
294            != self.connection.revocation.load(Ordering::Acquire)
295        {
296            status.valid_until = None;
297            if matches!(
298                status.state,
299                RegistrationState::Active | RegistrationState::PendingHealth
300            ) {
301                status.state = RegistrationState::Suspended;
302                status.error = Some(DiscoveryError::new(
303                    ErrorCode::Unauthenticated,
304                    "registration evidence belongs to a revoked identity",
305                ));
306            }
307        }
308        if !self.spec.eligibility.is_ready() || gate.failed || gate.close_started {
309            status.valid_until = None;
310        }
311        status
312    }
313}
314
315struct Admission<'a> {
316    core: &'a Core,
317    record: &'a Record,
318}
319impl WriteAdmission for Admission<'_> {
320    /// 业务作用:在 adapter 的实际传输提交点检查发布权并保存未知结果事实。
321    ///
322    /// 参数说明:
323    /// - `submit`:是已经完成鉴权准备的同步提交动作。
324    ///
325    /// 返回:许可有效时启动写入,失权时拒绝且不执行动作。
326    fn submit(&self, submit: &mut dyn FnMut()) -> Result<()> {
327        let (core, record) = (self.core, self.record);
328        let gate = lock(&core.gate);
329        let eligible = lock(&record.spec.eligibility.inner);
330        if gate.failed
331            || gate.close_started
332            || !matches!(gate.phase, OwnerPhase::Starting | OwnerPhase::Serving)
333            || !record.spec.eligibility.current(*eligible)
334            || core.registrations_stop.is_cancelled()
335        {
336            return Err(closed());
337        }
338        // 写前保留结果不明事实;等待 future 被取消时清理仍知道可能已经写入。
339        record.attempted.store(true, Ordering::Release);
340        record.uncertain.store(true, Ordering::Release);
341        submit();
342        Ok(())
343    }
344}
345
346/// 业务作用:每个注册目标只运行一个写入与健康监督序列。
347///
348/// 参数说明:
349/// - `core`:管理停止信号
350/// - `record`:保存不可变外部身份。
351///
352/// 返回:关闭后退出;已确认身份的幂等续约可协调重试,管理端禁用、缺失及未解决的相反写入不授予重新注册权限。
353pub(crate) async fn run(core: Arc<Core>, record: Arc<Record>) {
354    let mut eligibility = record.spec.eligibility.changes.subscribe();
355    let mut prepared = record.connection.prepared.subscribe();
356    let admission = Admission {
357        core: &core,
358        record: &record,
359    };
360    let monitor = millis(record.spec.policy.monitor_interval_ms);
361    let interval = record
362        .connection
363        .provider
364        .renewal_interval(&record.spec.plan)
365        .unwrap_or(monitor)
366        .min(monitor);
367    let idempotent_renewal = record
368        .connection
369        .provider
370        .supports_idempotent_renewal(&record.spec.plan);
371    let mut registered = false;
372    let mut renewal_allowed = false;
373    let mut registration_blocked = false;
374    let mut backoff = record.connection.policy.initial_backoff_ms;
375    let mut attempt = 0u64;
376    loop {
377        let mut renewal_failed = false;
378        if core.registrations_stop.is_cancelled() {
379            return;
380        }
381        if !record.spec.eligibility.is_ready() || !*prepared.borrow() {
382            if registered {
383                // 本地失权先停止健康发布;只有本 owner 明确完成的删除才允许资格恢复后重建。
384                set_state(&core, &record, RegistrationState::Suspended, None);
385                withdraw(
386                    record.clone(),
387                    Instant::now() + millis(record.spec.policy.deregister_timeout_ms),
388                    millis(record.spec.policy.deregister_timeout_ms),
389                )
390                .await;
391                registered = false;
392                registration_blocked |= lock(&record.status).state != RegistrationState::Removed;
393                changed(&core);
394            }
395            tokio::select! {
396                _ = core.registrations_stop.cancelled() => return,
397                _ = eligibility.changed() => {},
398                _ = prepared.changed() => {},
399                _ = tokio::time::sleep(interval) => {},
400            }
401            continue;
402        }
403        if !registered && !registration_blocked {
404            set_state(&core, &record, RegistrationState::Registering, None);
405            let result = operation(
406                &core,
407                Instant::now() + millis(record.connection.policy.request_timeout_ms),
408                record
409                    .connection
410                    .provider
411                    .register(&record.spec.plan, &admission),
412            )
413            .await;
414            match result {
415                Ok(()) => {
416                    record.uncertain.store(false, Ordering::Release);
417                    registered = true;
418                    renewal_allowed = true;
419                    set_state(&core, &record, RegistrationState::PendingHealth, None);
420                }
421                Err(error) => {
422                    {
423                        let gate = lock(&core.gate);
424                        revoke_identity(&core, &gate, &record.connection, &error);
425                    }
426                    if definite_rejection(&error) {
427                        record.uncertain.store(false, Ordering::Release);
428                    }
429                    registration_blocked = record.uncertain.load(Ordering::Acquire)
430                        || matches!(
431                            error.code,
432                            ErrorCode::InvalidConfig | ErrorCode::Unsupported
433                        );
434                    // 首次占位冲突可能来自旧 owner 的延迟删除;只允许重新核对,不授予覆盖既有记录的权限。
435                    set_state(
436                        &core,
437                        &record,
438                        if record.uncertain.load(Ordering::Acquire) {
439                            RegistrationState::OutcomeUnknown
440                        } else {
441                            RegistrationState::Suspended
442                        },
443                        Some(error),
444                    );
445                }
446            }
447        }
448        if registered
449            && renewal_allowed
450            // 只有已确认身份且 provider 承诺相同 payload 幂等时,才可协调未知续约;不重放首次注册。
451            && (!record.uncertain.load(Ordering::Acquire) || idempotent_renewal)
452            && record
453                .connection
454                .provider
455                .renewal_interval(&record.spec.plan)
456                .is_some()
457        {
458            let result = operation(
459                &core,
460                Instant::now() + millis(record.connection.policy.request_timeout_ms),
461                record
462                    .connection
463                    .provider
464                    .renew(&record.spec.plan, true, &admission),
465            )
466            .await;
467            match result {
468                // 本次续约确认后才恢复健康回查;更早未决续约的清理限制由独立事实保留。
469                Ok(()) => record.uncertain.store(false, Ordering::Release),
470                Err(error) => {
471                    renewal_failed = true;
472                    {
473                        let gate = lock(&core.gate);
474                        revoke_identity(&core, &gate, &record.connection, &error);
475                    }
476                    if definite_rejection(&error) {
477                        record.uncertain.store(false, Ordering::Release);
478                    }
479                    if record.uncertain.load(Ordering::Acquire) {
480                        // 新续约成功只能证明当前同向状态;更早请求仍可能迟到,注销及后续重建必须保留此事实。
481                        record.unsettled_renewal.store(true, Ordering::Release);
482                        renewal_allowed = idempotent_renewal && !error.revokes_evidence();
483                        set_state(
484                            &core,
485                            &record,
486                            RegistrationState::OutcomeUnknown,
487                            Some(error),
488                        );
489                    } else {
490                        renewal_allowed = false;
491                        let mut status = lock(&record.status);
492                        if error.revokes_evidence() {
493                            status.valid_until = None;
494                            status.state = RegistrationState::Suspended;
495                        }
496                        status.error = Some(error);
497                    }
498                }
499            }
500        }
501        if registered && !record.uncertain.load(Ordering::Acquire) {
502            let revocation = record.connection.revocation.load(Ordering::Acquire);
503            let result = operation(
504                &core,
505                Instant::now() + millis(record.connection.policy.request_timeout_ms),
506                record.connection.provider.inspect(&record.spec.plan),
507            )
508            .await;
509            let gate = lock(&core.gate);
510            if core.registrations_stop.is_cancelled() || gate.close_started {
511                return;
512            }
513            if let Err(error) = &result {
514                if error.revokes_evidence() {
515                    renewal_allowed = false;
516                }
517                revoke_identity(&core, &gate, &record.connection, error);
518            }
519            // 回查期间发生撤权时,旧请求的 Healthy 或其他结果都不能发布;等待原身份下的新一轮证据。
520            if revocation == record.connection.revocation.load(Ordering::Acquire) {
521                let mut status = lock(&record.status);
522                match result {
523                    Ok(RegistrationEvidence::Healthy) if record.spec.eligibility.is_ready() => {
524                        registration_blocked = false;
525                        renewal_allowed = true;
526                        let now = Instant::now();
527                        status.state = RegistrationState::Active;
528                        status.observed_at = Some(now);
529                        status.valid_until =
530                            Some(now + millis(record.spec.policy.observation_grace_ms));
531                        status.error = None;
532                        status.startup_confirmed = true;
533                        record.confirmed.store(true, Ordering::Release);
534                        record
535                            .health_revocation
536                            .store(revocation, Ordering::Release);
537                    }
538                    Ok(RegistrationEvidence::Disabled | RegistrationEvidence::Missing) => {
539                        // 管理端移除和禁用只允许继续观察,不能据此取得重新注册的权限。
540                        renewal_allowed = false;
541                        registration_blocked = true;
542                        status.state = RegistrationState::Suspended;
543                        status.valid_until = None;
544                        status.error = Some(DiscoveryError::new(
545                            ErrorCode::NotReady,
546                            "registration was disabled or removed",
547                        ));
548                    }
549                    Ok(_) => {
550                        renewal_allowed = true;
551                        status.state = RegistrationState::PendingHealth;
552                        status.valid_until = None;
553                        status.error = None;
554                    }
555                    Err(error) => {
556                        // 不完整回查不产生新的健康事实,保留最后一次 Healthy 的固定期限;明确撤权仍立即关闭注册资格。
557                        if error.revokes_evidence() {
558                            status.state = RegistrationState::Suspended;
559                            status.valid_until = None;
560                            renewal_allowed = false;
561                        }
562                        status.error = Some(error);
563                    }
564                }
565            }
566        }
567        changed(&core);
568        let delay = if registered && !renewal_failed {
569            backoff = record.connection.policy.initial_backoff_ms;
570            attempt = 0;
571            interval
572        } else {
573            let delay = crate::retry_delay(&core, &record.connection, backoff, attempt);
574            attempt = attempt.wrapping_add(1);
575            backoff = backoff
576                .saturating_mul(2)
577                .min(record.connection.policy.max_backoff_ms);
578            // 续约响应持续失败也要退避,且不能比正常健康上报周期更频繁地重发。
579            delay.max(interval)
580        };
581        let next_attempt = Instant::now() + delay;
582        loop {
583            tokio::select! {
584                _ = core.registrations_stop.cancelled() => return,
585                // 持续刷新 true 不能绕过退避;实际失权立即唤醒,以便停止续约并执行摘流。
586                _ = eligibility.changed() => if !record.spec.eligibility.is_ready() { break; },
587                _ = tokio::time::sleep_until(next_attempt) => break,
588            }
589        }
590    }
591}
592
593/// 业务作用:注册或健康回查的认证撤销必须同时关闭该身份下的全部发现路由。
594///
595/// 参数说明:
596/// - `core`:提供通知
597/// - `gate`:是已持有的状态门禁
598/// - `connection`:是明确连接
599/// - `error`:是后端稳定错误类别。
600///
601/// 返回:只有身份或权限失败会撤销连接证据,单注册冲突不影响其他服务。
602fn revoke_identity(core: &Core, gate: &Gate, connection: &Connection, error: &DiscoveryError) {
603    if matches!(
604        error.code,
605        ErrorCode::Unauthenticated | ErrorCode::PermissionDenied | ErrorCode::PermissionFiltered
606    ) {
607        crate::revoke_connection(core, gate, connection);
608    }
609}
610
611/// 业务作用:区分明确拒绝与请求可能已在远端执行的故障。
612///
613/// 参数说明:
614/// - `error`:由 adapter 将服务端拒绝映射为稳定类别。
615///
616/// 返回:仅明确未执行的拒绝允许清除本次写入的不确定状态。
617fn definite_rejection(error: &DiscoveryError) -> bool {
618    matches!(
619        error.code,
620        ErrorCode::InvalidConfig
621            | ErrorCode::Unauthenticated
622            | ErrorCode::PermissionDenied
623            | ErrorCode::Conflict
624            | ErrorCode::Unsupported
625    )
626}
627
628/// 业务作用:发布注册状态变化,不在状态锁内执行后端调用。
629///
630/// 参数说明:
631/// - `core`:提供全局状态变化通知的运行时核心。
632/// - `record`:需要更新本地注册状态的记录。
633/// - `state`:当前裁决得到的注册状态。
634/// - `error`:与本次状态一并发布的可选受控错误。
635///
636/// 返回:无返回值;锁中更新状态和错误,非 Active 时清空 valid_until,随后解锁并通知变化。Active 时保留已有有效期,不自行续租或调用后端。
637fn set_state(
638    core: &Core,
639    record: &Record,
640    state: RegistrationState,
641    error: Option<DiscoveryError>,
642) {
643    let mut status = lock(&record.status);
644    status.state = state;
645    status.error = error;
646    if state != RegistrationState::Active {
647        status.valid_until = None;
648    }
649    drop(status);
650    changed(core);
651}
652
653/// 业务作用:尝试删除本 owner 曾提交的注册,结果不明时保留残留事实。
654///
655/// 参数说明:
656/// - `record`:是精确计划
657/// - `deadline`:是全局期限
658/// - `share`:防止单目标耗尽全部预算。
659///
660/// 返回:没有写过或明确删除且无旧未知写时标记 Removed,其余标记 Residual。
661pub(crate) async fn withdraw(record: Arc<Record>, deadline: Instant, share: Duration) {
662    if !record.attempted.load(Ordering::Acquire) {
663        lock(&record.status).state = RegistrationState::Removed;
664        return;
665    }
666    if lock(&record.status).state == RegistrationState::Removed {
667        return;
668    }
669    // 后续同向续约或普通回查不能证明旧请求已终止,清理 ACK 不得清除其迟到风险或开放重新注册。
670    let uncertain = record.uncertain.load(Ordering::Acquire)
671        || record.unsettled_renewal.load(Ordering::Acquire);
672    lock(&record.status).state = RegistrationState::Withdrawing;
673    let until =
674        deadline.min(Instant::now() + share.min(millis(record.spec.policy.deregister_timeout_ms)));
675    let result = tokio::time::timeout_at(
676        until,
677        record.connection.provider.deregister(&record.spec.plan),
678    )
679    .await
680    .unwrap_or_else(|_| Err(timeout()));
681    let mut status = lock(&record.status);
682    status.valid_until = None;
683    match result {
684        Ok(()) if !uncertain => {
685            status.state = RegistrationState::Removed;
686            status.error = None;
687            record.attempted.store(false, Ordering::Release);
688        }
689        Ok(()) => {
690            status.state = RegistrationState::Residual;
691            status.error = Some(DiscoveryError::new(
692                ErrorCode::OutcomeUnknown,
693                "an earlier write may still complete after cleanup",
694            ));
695        }
696        Err(error) => {
697            record.uncertain.store(true, Ordering::Release);
698            status.state = RegistrationState::Residual;
699            status.error = Some(error);
700        }
701    }
702}
703
704/// 业务作用:摘流终止后保留不能证明已删除的外部记录。
705///
706/// 参数说明:
707/// - `record`:的写入和摘流任务均已结束。
708///
709/// 返回:未写入的目标为 Removed,其余未完成目标为 Residual。
710pub(crate) fn finish_withdrawal(record: &Record) {
711    let mut status = lock(&record.status);
712    status.valid_until = None;
713    if !record.attempted.load(Ordering::Acquire) {
714        status.state = RegistrationState::Removed;
715    } else if !matches!(
716        status.state,
717        RegistrationState::Removed | RegistrationState::Residual
718    ) {
719        status.state = RegistrationState::Residual;
720        status.error = Some(DiscoveryError::new(
721            ErrorCode::OutcomeUnknown,
722            "registration cleanup did not complete",
723        ));
724    }
725}