diff --git a/crates/core/src/chain/backends.rs b/crates/core/src/chain/backends.rs index 11d93cf3..88f55212 100644 --- a/crates/core/src/chain/backends.rs +++ b/crates/core/src/chain/backends.rs @@ -8707,6 +8707,62 @@ impl ChainBackend for RealSellerBackend { ))) } + async fn seller_offer_outcome_since( + &self, + tc: &TokenContract, + since: u64, + ) -> Result, ChainError> { + // This path is used by a durable submission marker after a restart. + // Do not read `offer_post_started_at` here: it belongs to the former + // process and therefore cannot establish an event boundary safely. + let ob = retry_seller_read("seller marker outcome order-book address", || async { + self.chain + .inference_orderbook_address(&self.note, &self.model_hash, self.tick_size) + .await + .map_err(map_err) + }) + .await?; + let tc_addr = parse_tc(tc)?; + let events = retry_seller_read("seller marker-bounded outcome events", || async { + self.chain + .seller_offer_events_since(&self.note, &ob, &tc_addr, since) + .await + .map_err(map_err) + }) + .await?; + let matched_state = retry_seller_read("seller marker immediate-match state", || async { + self.read_openable_match_once(tc).await + }) + .await? + .is_some(); + match classify_seller_offer_outcome(events, matched_state) { + Ok(outcome) => Ok(outcome), + Err(ChainError::DuplicateSell(_)) => { + let latch = + retry_seller_read("seller marker TokenContract offer latch", || async { + self.chain + .token_contract_offer(&tc_addr) + .await + .map_err(map_err) + }) + .await?; + Err(duplicate_sell_from_offer_latch(&tc_addr, latch)) + } + Err(other) => Err(other), + } + } + + async fn seller_offer_latch( + &self, + token_contract: &TokenContract, + ) -> Result, ChainError> { + let tc = parse_tc(token_contract)?; + retry_seller_read("seller TokenContract offer latch", || async { + self.chain.token_contract_offer(&tc).await.map_err(map_err) + }) + .await + } + async fn sell_offer_terms( &self, token_contract: &TokenContract, diff --git a/crates/core/src/market/mod.rs b/crates/core/src/market/mod.rs index 3a36a3e1..2e51fe2a 100644 --- a/crates/core/src/market/mod.rs +++ b/crates/core/src/market/mod.rs @@ -160,6 +160,33 @@ pub trait ChainBackend: Send + Sync { ) -> Result, ChainError> { Ok(None) } + /// Read the owner-note placement/match facts for one exact TokenContract + /// from a durable event lower bound. This is intentionally separate from + /// `confirm_offer_outcome`: reconciliation after a service restart cannot + /// depend on a backend-local "post started" timestamp that vanished with + /// the old process. A backend which cannot make this authoritative read + /// must return an error so a persisted seller marker remains fail-closed. + async fn seller_offer_outcome_since( + &self, + _token_contract: &TokenContract, + _since_unix: u64, + ) -> Result, ChainError> { + Err(ChainError::Chain( + "marker-bounded seller offer event reconciliation is not supported by this backend" + .to_string(), + )) + } + /// Read the exact deal's post-offer latch. This is deliberately separate from a + /// book row: a successful `postSellOffer` can be accepted before an index/read + /// path catches up, and a returned placement value is not by itself a proof that + /// a successor post is safe. Backends which cannot read the live deal return + /// `None`; a persisted real-seller submission must then remain unconfirmed. + async fn seller_offer_latch( + &self, + _token_contract: &TokenContract, + ) -> Result, ChainError> { + Ok(None) + } /// Read the authoritative sell-offer terms for a real per-deal `TokenContract`. The real seller path uses /// this before posting an ask so CLI defaults/prompts cannot diverge from the already-deployed TC config. /// Mock backends have no on-chain TC config, so they return `None`. diff --git a/crates/dexdo/src/cli/args.rs b/crates/dexdo/src/cli/args.rs index cd528e43..a9fe9c58 100644 --- a/crates/dexdo/src/cli/args.rs +++ b/crates/dexdo/src/cli/args.rs @@ -281,6 +281,15 @@ pub(crate) struct SellerArgs { /// to the legacy XDG/Windows config path. Real seller startup fails closed if missing or incomplete. #[arg(long)] pub(crate) policy: Option, + /// Recovery escape hatch for a retained publication marker. The value must + /// exactly name the selected TokenContract and is accepted only with the + /// separate confirmation flag; service/timer units must never set it. + #[arg(long, value_name = "TOKEN_CONTRACT")] + pub(crate) recover_publication: Option, + /// Acknowledge that `--recover-publication` creates an immutable audit and + /// permits exactly one fresh explicit seller post after exact-negative proof. + #[arg(long, requires = "recover_publication")] + pub(crate) confirm_recover_publication: bool, } impl SellerArgs { diff --git a/crates/dexdo/src/cli/seller.rs b/crates/dexdo/src/cli/seller.rs index 562fed74..dd626bcd 100644 --- a/crates/dexdo/src/cli/seller.rs +++ b/crates/dexdo/src/cli/seller.rs @@ -1232,6 +1232,9 @@ struct SellerPoolContext<'a> { gateway_advertise: &'a str, /// how a failed `advertised_gateway` self-probe is treated. advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy, + /// Present only for a human-invoked seller command which supplied both + /// explicit recovery flags. Timer/controller startup constructs `None`. + recover_publication: Option<&'a str>, } fn save_pool_deal_handle(context: &SellerPoolContext<'_>, deal: &SellerPoolDeal) -> Result<()> { @@ -2248,17 +2251,37 @@ where dexdo::seller::SellerOfferInspection::Funded | dexdo::seller::SellerOfferInspection::Vacant => None, }; - let startup = match dexdo::seller::liveness::prepare_seller_offer_with_liveness( - seller, - deal.chain.as_ref(), - &deal.cfg, - context.note_addr, - inspected_identity.as_ref(), - shutdown.as_mut(), - context.advertise_probe, - ) - .await? - { + let manual_recovery = context.recover_publication.is_some_and(|token_contract| { + token_contract.eq_ignore_ascii_case(&deal.cfg.token_contract) + }); + let startup = match if manual_recovery { + dexdo::seller::liveness::prepare_seller_offer_with_audited_manual_recovery_liveness( + seller, + deal.chain.as_ref(), + &deal.cfg, + context.note_addr, + inspected_identity.as_ref(), + &deal.watch.cursor_path, + context + .recover_publication + .expect("manual recovery was checked"), + shutdown.as_mut(), + context.advertise_probe, + ) + .await + } else { + dexdo::seller::liveness::prepare_seller_offer_with_persisted_liveness( + seller, + deal.chain.as_ref(), + &deal.cfg, + context.note_addr, + inspected_identity.as_ref(), + &deal.watch.cursor_path, + shutdown.as_mut(), + context.advertise_probe, + ) + .await + }? { dexdo::seller::liveness::SellerStartupOutcome::Ready(startup) => { // shutdown can win the startup select, then an already-matched SELL turns the // result back into `Ready`. The completed `Fuse` is the retained witness that this @@ -3720,6 +3743,33 @@ pub(crate) async fn run_seller_with_deal_gas_overhead( // The manifest's frame_model (if any) is validated against `--model` inside `seller_real_backend`. let (mut token_contract, mut market_frame_model, market_nonce) = resolve_market_fields(args.market.as_deref(), args.token_contract.as_deref(), None)?; + match ( + args.recover_publication.as_deref(), + args.confirm_recover_publication, + ) { + (Some(_), false) => bail!( + "--recover-publication requires --confirm-recover-publication; automatic service/timer restarts must not use this escape hatch" + ), + (Some(operator_tc), true) => { + // A market manifest and a human command may name the same account in + // different accepted address representations. Compare the parsed + // account, not its spelling; the recovery still binds to the selected + // manifest token contract below. + let operator_tc = dexdo_core::normalize_wallet_address(operator_tc).map_err(|error| { + anyhow::anyhow!("--recover-publication has an invalid TokenContract: {error}") + })?; + let selected_tc = dexdo_core::normalize_wallet_address(&token_contract).map_err(|error| { + anyhow::anyhow!("selected TokenContract is invalid: {error}") + })?; + if operator_tc != selected_tc { + bail!( + "--recover-publication does not match the selected TokenContract" + ); + } + } + (None, true) => unreachable!("clap requires --recover-publication"), + _ => {} + } let mut startup_market = args.market.as_deref().map(load_market).transpose()?; // Review: the deal nonce comes from `--market` (the manifest) or the explicit `--nonce` flag -- // never both (the manifest is the single source of truth). The real-chain seller path requires @@ -4420,6 +4470,13 @@ pub(crate) async fn run_seller_with_deal_gas_overhead( frame_model, gateway_advertise: &gateway_advertise, advertise_probe: args.advertise_probe_policy(), + // The operator's spelling was checked above. Pass the selected runtime + // identity forward so the audited one-shot permit uses precisely the + // token contract that this seller will serve. + recover_publication: args + .recover_publication + .as_deref() + .map(|_| token_contract.as_str()), }; if !args.mock.mock_chain { sweep_configured_seller_model_books(&args, note_addr, context.frame_model, &token_contract) @@ -4671,6 +4728,8 @@ mod tests { allow_unverified_model: false, models: root.path().join("missing-models.json"), policy: Some(policy_path), + recover_publication: None, + confirm_recover_publication: false, }) .await .expect_err("second seller process for one note must fail on the production lock"); @@ -5135,6 +5194,7 @@ mod tests { frame_model: "mock", gateway_advertise: &config.gateway_advertise, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }; let mut provisioner = |_: String, _: u64, _: u64, _: u64| { futures::future::ready(Err::< @@ -5437,6 +5497,7 @@ mod tests { frame_model: "mock", gateway_advertise: &gateway_advertise, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }; let shutdown = futures::future::pending::<()>(); tokio::pin!(shutdown); @@ -5734,6 +5795,8 @@ mod tests { allow_unverified_model: false, models: root.join("models.json"), policy: None, + recover_publication: None, + confirm_recover_publication: false, }) .await; let error = match case { @@ -6585,6 +6648,7 @@ mod tests { frame_model, gateway_advertise: &gateway, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }, &pool_test_policy(2), &mut provision, @@ -6843,6 +6907,7 @@ mod tests { frame_model: "openai/gpt-oss-20b", gateway_advertise: &gateway, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }, &pool_test_policy(1), &mut provision, @@ -7349,6 +7414,7 @@ mod tests { frame_model: "openai/gpt-oss-20b", gateway_advertise: &gateway, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }, false, shutdown.as_mut(), @@ -7495,6 +7561,7 @@ mod tests { frame_model, gateway_advertise: &gateway, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }, &pool_test_policy(3), &mut provision, @@ -7566,6 +7633,7 @@ mod tests { frame_model, gateway_advertise: gateway, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, } } @@ -8097,6 +8165,7 @@ mod tests { // `unreachable` is a closed loopback port, so it is not public and the // production default is still fatal -- the cascade under test is reached. advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, }, &pool_test_policy(2), &mut provision, @@ -8198,6 +8267,8 @@ mod tests { allow_unverified_model: false, models: root.path().join("unused-models.json"), policy: None, + recover_publication: None, + confirm_recover_publication: false, }; assert_eq!( args.checked_gateway_advertise_addr().unwrap(), @@ -8393,6 +8464,8 @@ mod tests { allow_unverified_model: false, models: root.path().join("unused-models.json"), policy: None, + recover_publication: None, + confirm_recover_publication: false, }; let seller = super::run_seller(args); @@ -8591,6 +8664,7 @@ mod tests { frame_model: "mock", gateway_advertise: &advertise, advertise_probe: args.advertise_probe_policy(), + recover_publication: None, }, &pool_test_policy(1), &mut provision, @@ -8697,6 +8771,7 @@ mod tests { frame_model: "mock", gateway_advertise: &advertise, advertise_probe: args.advertise_probe_policy(), + recover_publication: None, }, &pool_test_policy(1), &mut provision, @@ -8752,6 +8827,8 @@ mod tests { allow_unverified_model: false, models: root.join("unused-models.json"), policy: None, + recover_publication: None, + confirm_recover_publication: false, } } diff --git a/crates/dexdo/src/cli/seller_1056_restart_tests.rs b/crates/dexdo/src/cli/seller_1056_restart_tests.rs index 2983480e..ccdaa1d3 100644 --- a/crates/dexdo/src/cli/seller_1056_restart_tests.rs +++ b/crates/dexdo/src/cli/seller_1056_restart_tests.rs @@ -166,6 +166,8 @@ fn issue_1056_seller_args( allow_unverified_model: false, models: root.join("unused-models.json"), policy: None, + recover_publication: None, + confirm_recover_publication: false, } } diff --git a/crates/dexdo/src/cli/seller_1057_shutdown_tests.rs b/crates/dexdo/src/cli/seller_1057_shutdown_tests.rs index 2115ff34..2f96e016 100644 --- a/crates/dexdo/src/cli/seller_1057_shutdown_tests.rs +++ b/crates/dexdo/src/cli/seller_1057_shutdown_tests.rs @@ -18,6 +18,7 @@ impl Issue1057Pool { frame_model: self.frame_model, gateway_advertise: &self.gateway, advertise_probe: dexdo::seller::liveness::AdvertiseProbePolicy::default(), + recover_publication: None, } } } diff --git a/crates/dexdo/src/cli/seller_1402_refusal_tests.rs b/crates/dexdo/src/cli/seller_1402_refusal_tests.rs index 9cfec342..a5393900 100644 --- a/crates/dexdo/src/cli/seller_1402_refusal_tests.rs +++ b/crates/dexdo/src/cli/seller_1402_refusal_tests.rs @@ -51,6 +51,8 @@ fn refusal_seller_args( allow_unverified_model: false, models: root.join("unused-models.json"), policy: Some(policy), + recover_publication: None, + confirm_recover_publication: false, } } diff --git a/crates/dexdo/src/seller/liveness.rs b/crates/dexdo/src/seller/liveness.rs index fcc6d6b3..2bb91d95 100644 --- a/crates/dexdo/src/seller/liveness.rs +++ b/crates/dexdo/src/seller/liveness.rs @@ -1,8 +1,11 @@ use super::{ - inspect_seller_offer, prepare_seller_offer, validate_resting_offer, wait_for_match, - RunningSeller, SellerConfig, SellerMatchWatchConfig, SellerOfferInspection, SellerOfferStartup, + authorize_exact_negative_manual_recovery, clear_offer_submission, + consume_exact_negative_manual_recovery, inspect_seller_offer, pending_offer_submission, + persist_offer_submission, prepare_seller_offer, reconcile_pending_offer_submission, + validate_resting_offer, wait_for_match, RunningSeller, SellerConfig, SellerMatchWatchConfig, + SellerOfferInspection, SellerOfferStartup, }; -use anyhow::Result; +use anyhow::{anyhow, Result}; use dexdo_core::{ market::{RestingSellCancelStartError, RestingSellCancelWatch}, params::{ @@ -13,6 +16,7 @@ use dexdo_core::{ }; use dexdo_proto::{ChallengeRequest, GatewayClient}; use std::future::Future; +use std::path::Path; use std::time::Duration; use crate::seller::auth::HEALTH_CHALLENGE_TC; @@ -2013,21 +2017,25 @@ where } Ok(startup) => Ok(SellerStartupOutcome::Ready(startup)), Err(error) => { + // A post/confirmation error is not proof that the fresh + // write failed. In particular, cancelling a row which + // becomes visible during this error path turns a + // transport timeout into an unauthorized lifecycle + // action. Leave it fail-closed for the durable marker + // reconciler instead; it is the only path allowed to + // adopt a later exact-TC fact or permit a future retry. let reason = RestingStopReason::Watcher(format!( - "seller offer post or confirmation failed: {error}" + "publication_unconfirmed: seller offer post or confirmation failed: {error}" )); - resolve_interrupted_startup( - seller, - chain, - cfg, - expected_owner, - None, - ( - reason, - tokio::time::Instant::now() + timing.cycle_timeout, - ), - timing, - ).await + Ok(SellerStartupOutcome::Stopped { + identity: None, + reason, + disposition: CancellationDisposition::UnknownFailure { + known_result: format!( + "publication_unconfirmed; no cancellation or repost was submitted after seller post/confirmation error: {error}" + ), + }, + }) } }; } @@ -2087,6 +2095,174 @@ where .await } +/// Production-only write-ahead wrapper around the normal liveness startup. +/// +/// The existing liveness path owns readiness, cancellation and gateway +/// supervision. This wrapper adds one invariant at its outer money boundary: +/// after a marker exists, the next start first reconciles the exact deal and +/// never falls through to another post merely because a read timed out. +pub async fn prepare_seller_offer_with_persisted_liveness( + seller: &RunningSeller, + chain: &dyn ChainBackend, + cfg: &SellerConfig, + expected_owner: &str, + existing_identity: Option<&RestingOfferIdentity>, + cursor_path: &Path, + shutdown: S, + advertise_probe: AdvertiseProbePolicy, +) -> Result +where + S: Future, +{ + if let Some(marker) = pending_offer_submission(cursor_path, &cfg.token_contract)? { + match reconcile_pending_offer_submission(chain, cfg, Some(expected_owner), &marker).await { + Ok(super::PendingPublicationResolution::Resting { order_id }) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + return Ok(SellerStartupOutcome::Ready( + SellerOfferStartup::ResumedResting { order_id }, + )); + } + Ok(super::PendingPublicationResolution::Funded) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + return Ok(SellerStartupOutcome::Ready( + SellerOfferStartup::ResumedFunded, + )); + } + Ok(super::PendingPublicationResolution::Negative) => { + return Err(anyhow!( + "publication_unconfirmed token_contract={}: the prior marked post has an exact negative proof; marker is retained so an automatic restart cannot retry — use an audited manual recovery operation", + display_token_contract(&cfg.token_contract) + )); + } + Err(error) => return Err(error), + } + } else { + // This write is intentionally before liveness enters the only path + // which can call postSellOffer. If readiness aborts before the write, + // the next explicit start obtains a harmless exact negative proof; the + // conservative false-positive is preferable to a duplicate SELL. + persist_offer_submission(cursor_path, &cfg.token_contract)?; + } + + let started = prepare_seller_offer_with_liveness( + seller, + chain, + cfg, + expected_owner, + existing_identity, + shutdown, + advertise_probe, + ) + .await; + + let marker = pending_offer_submission(cursor_path, &cfg.token_contract)?.ok_or_else(|| { + anyhow!( + "publication_unconfirmed token_contract={}: durable marker disappeared before postSellOffer reconciliation", + display_token_contract(&cfg.token_contract) + ) + })?; + match reconcile_pending_offer_submission(chain, cfg, Some(expected_owner), &marker).await { + Ok(super::PendingPublicationResolution::Resting { order_id }) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + match started { + Ok(SellerStartupOutcome::Stopped { .. }) => Ok(SellerStartupOutcome::Ready( + SellerOfferStartup::ResumedResting { order_id }, + )), + other => other, + } + } + Ok(super::PendingPublicationResolution::Funded) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + match started { + Ok(SellerStartupOutcome::Stopped { .. }) => Ok(SellerStartupOutcome::Ready( + SellerOfferStartup::ResumedFunded, + )), + other => other, + } + } + Ok(super::PendingPublicationResolution::Negative) => { + Err(anyhow!( + "publication_unconfirmed token_contract={}: the marked post has an exact negative proof; marker is retained so an automatic restart cannot retry — use an audited manual recovery operation", + display_token_contract(&cfg.token_contract) + )) + } + Err(error) => Err(error), + } +} + +/// The only route that may make a fresh post after a retained marker has an +/// exact negative proof. Its caller must be an explicit operator command; +/// ordinary liveness/controller startup always uses the wrapper above and +/// remains unable to consume this permit. +pub async fn prepare_seller_offer_with_audited_manual_recovery_liveness( + seller: &RunningSeller, + chain: &dyn ChainBackend, + cfg: &SellerConfig, + expected_owner: &str, + existing_identity: Option<&RestingOfferIdentity>, + cursor_path: &Path, + operator_token_contract: &str, + shutdown: S, + advertise_probe: AdvertiseProbePolicy, +) -> Result +where + S: Future, +{ + authorize_exact_negative_manual_recovery( + chain, + cfg, + expected_owner, + cursor_path, + operator_token_contract, + ) + .await?; + // Persist this one-shot transition before `prepare_seller_offer_with_liveness` + // reaches its only post path. A process crash here is conservative: another + // automatic run sees the consumed marker and remains blocked. + consume_exact_negative_manual_recovery(cursor_path, &cfg.token_contract)?; + let started = prepare_seller_offer_with_liveness( + seller, + chain, + cfg, + expected_owner, + existing_identity, + shutdown, + advertise_probe, + ) + .await; + let marker = pending_offer_submission(cursor_path, &cfg.token_contract)?.ok_or_else(|| { + anyhow!( + "publication_unconfirmed token_contract={}: audited manual recovery marker disappeared before reconciliation", + display_token_contract(&cfg.token_contract) + ) + })?; + match reconcile_pending_offer_submission(chain, cfg, Some(expected_owner), &marker).await { + Ok(super::PendingPublicationResolution::Resting { order_id }) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + match started { + Ok(SellerStartupOutcome::Stopped { .. }) => Ok(SellerStartupOutcome::Ready( + SellerOfferStartup::ResumedResting { order_id }, + )), + other => other, + } + } + Ok(super::PendingPublicationResolution::Funded) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + match started { + Ok(SellerStartupOutcome::Stopped { .. }) => Ok(SellerStartupOutcome::Ready( + SellerOfferStartup::ResumedFunded, + )), + other => other, + } + } + Ok(super::PendingPublicationResolution::Negative) => Err(anyhow!( + "publication_unconfirmed token_contract={}: audited manual recovery post has an exact negative proof; the consumed marker is retained and no automatic retry is permitted", + display_token_contract(&cfg.token_contract) + )), + Err(error) => Err(error), + } +} + #[derive(Clone, Copy)] struct SupervisionTiming { health_interval: Duration, @@ -3004,6 +3180,39 @@ mod tests { })) } + async fn seller_offer_outcome_since( + &self, + token_contract: &TokenContract, + _: u64, + ) -> Result, ChainError> { + if self.matched.lock().unwrap().is_some() { + return Ok(Some(SellOfferOutcome::Matched)); + } + Ok(self + .orders + .lock() + .unwrap() + .iter() + .find(|order| order.token_contract.as_ref() == Some(token_contract)) + .map(|order| SellOfferOutcome::Rested { + order_id: order.order_id, + })) + } + + async fn seller_offer_latch( + &self, + token_contract: &TokenContract, + ) -> Result, ChainError> { + Ok(Some(DealOfferLatch { + offer_posted: self + .orders + .lock() + .unwrap() + .iter() + .any(|order| order.token_contract.as_ref() == Some(token_contract)), + })) + } + async fn raw_resting_sell_orders_for_tc( &self, token_contract: &TokenContract, @@ -3285,6 +3494,133 @@ mod tests { } } + #[tokio::test] + async fn persisted_liveness_negative_marker_blocks_an_automatic_service_restart() { + let seller = super::super::start_gateway_with_note( + "127.0.0.1:0".parse().unwrap(), + UpstreamConfig::Mock, + Arc::new(LocalNote::generate()), + ) + .await + .unwrap(); + let owner = address('a'); + let tc = address('b'); + let config = cfg_for_seller(&tc, &seller); + let backend = CancelBackend::new(Vec::new(), owner.clone(), 700, CancelBehavior::Remove); + let (_dir, watch) = watch("publication-negative-service-restart"); + persist_offer_submission(&watch.cursor_path, &tc).unwrap(); + + for attempt in 0..2 { + let error = prepare_seller_offer_with_persisted_liveness( + &seller, + &backend, + &config, + &owner, + None, + &watch.cursor_path, + std::future::pending(), + AdvertiseProbePolicy::default(), + ) + .await + .expect_err("negative marker must fence automatic service restart"); + assert!(error.to_string().contains("publication_unconfirmed")); + assert_eq!( + backend.posts.load(Ordering::Relaxed), + 0, + "attempt {attempt} posted despite the retained negative marker" + ); + assert!(pending_offer_submission(&watch.cursor_path, &tc) + .unwrap() + .is_some()); + } + seller.server_task.abort(); + } + + #[tokio::test] + async fn explicit_audited_recovery_is_the_only_liveness_route_that_reposts_a_negative_marker() { + let seller = super::super::start_gateway_with_note( + "127.0.0.1:0".parse().unwrap(), + UpstreamConfig::Mock, + Arc::new(LocalNote::generate()), + ) + .await + .unwrap(); + wait_for_gateway_ready(&seller).await; + let owner = address('a'); + let tc = address('b'); + let config = cfg_for_seller(&tc, &seller); + let backend = CancelBackend::new(Vec::new(), owner.clone(), 701, CancelBehavior::Remove); + let (_dir, watch) = watch("publication-explicit-manual-recovery"); + persist_offer_submission(&watch.cursor_path, &tc).unwrap(); + + let outcome = prepare_seller_offer_with_audited_manual_recovery_liveness( + &seller, + &backend, + &config, + &owner, + None, + &watch.cursor_path, + &tc, + std::future::pending(), + AdvertiseProbePolicy::default(), + ) + .await + .expect("explicit audited recovery may make exactly one replacement post"); + assert!(matches!(outcome, SellerStartupOutcome::Ready(_))); + assert_eq!(backend.posts.load(Ordering::Relaxed), 1); + assert!(pending_offer_submission(&watch.cursor_path, &tc) + .unwrap() + .is_none()); + seller.server_task.abort(); + } + + #[tokio::test] + async fn persisted_liveness_adopts_a_confirmed_exact_resting_sell_without_entering_post_path() { + let seller = super::super::start_gateway_with_note( + "127.0.0.1:0".parse().unwrap(), + UpstreamConfig::Mock, + Arc::new(LocalNote::generate()), + ) + .await + .unwrap(); + let owner = address('c'); + let tc = address('d'); + let config = cfg_for_seller(&tc, &seller); + let backend = CancelBackend::new( + vec![order(701, &owner, &tc)], + owner.clone(), + 701, + CancelBehavior::Remove, + ); + let (_dir, watch) = watch("publication-resting-service-restart"); + persist_offer_submission(&watch.cursor_path, &tc).unwrap(); + + let outcome = prepare_seller_offer_with_persisted_liveness( + &seller, + &backend, + &config, + &owner, + None, + &watch.cursor_path, + std::future::pending(), + AdvertiseProbePolicy::default(), + ) + .await + .unwrap(); + assert!( + matches!( + outcome, + SellerStartupOutcome::Ready(SellerOfferStartup::ResumedResting { order_id: 701 }) + ), + "unexpected persisted startup outcome: {outcome:?}" + ); + assert_eq!(backend.posts.load(Ordering::Relaxed), 0); + assert!(pending_offer_submission(&watch.cursor_path, &tc) + .unwrap() + .is_none()); + seller.server_task.abort(); + } + /// A resting SELL as the chain can actually hold one. The deadline is live and finite because a /// SELL's is mandatory: `PrivateNote.postSellOffer` refuses `ttl == 0` and caps it at /// `MAX_SELL_TTL = 3600` (`contracts/dex/PrivateNote.sol:41,792`), so a zero here would describe a diff --git a/crates/dexdo/src/seller/mod.rs b/crates/dexdo/src/seller/mod.rs index c7eb0dbd..fbc915a8 100644 --- a/crates/dexdo/src/seller/mod.rs +++ b/crates/dexdo/src/seller/mod.rs @@ -1,4 +1,4 @@ -//! Seller client: gateway + authorization + mock upstream + stream opening. +//! Seller client (§10.3, §10.5): gateway + authorization + mock upstream + stream opening. //! Headless (R12): starts without a GUI and serves the stream as a daemon. pub mod advance; @@ -32,6 +32,8 @@ use dexdo_core::{ SellOfferOutcome, TokenContract, SUBSCRIPTION_MAX_TICKS, SUBSCRIPTION_WEEKS, }; use gateway::{GatewayService, GatewayState}; +use sha2::{Digest, Sha256}; +use std::io::Write as _; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -42,17 +44,18 @@ use tonic::transport::{Identity, Server, ServerTlsConfig}; const SUBSCRIPTION_SELL_FLAGS: u8 = order_flags::AON | order_flags::SUBSCRIPTION; const SELLER_MATCH_WATCH_CURSOR_VERSION: u32 = 1; +const SELLER_OFFER_SUBMISSION_VERSION: u32 = 1; fn display_token_contract(token_contract: &str) -> String { dexdo_core::address::display_self_dapp(token_contract) } -/// Seller configuration for one stream. +/// Seller configuration for one stream (minimum for Directive 1). #[derive(Debug, Clone)] pub struct SellerConfig { - /// Contract -- the deal's handover point. + /// Contract — the deal's handover point (§2.1). pub token_contract: TokenContract, - /// Tick price `P` in raw ECC[2] units. Stated on the command line in whole SHELL + /// Tick price `P` in raw ECC[2] units (§1). Stated on the command line in whole SHELL /// (`--price-per-tick 3`) and converted once, at the argument. pub price_per_tick: u64, /// Maximum ticks in the offer. @@ -61,7 +64,7 @@ pub struct SellerConfig { pub subscription: bool, /// Public gateway host:port that will be encrypted to the buyer (R15). pub gateway_advertise: String, - /// How many fake tokens to yield (mock model). `0` = a deliberate seller no-show. + /// How many fake tokens to yield (mock model). `0` = a deliberate seller no-show (§3.1.2). /// Real upstreams are limited by the buyer request's `max_tokens` and the matched TC's strict /// `fundedTokens`/weekly cap, not by this debug fixture or the seller's advertised maximum. pub mock_token_count: u64, @@ -84,14 +87,14 @@ pub enum SellerOfferInspection { /// A running seller gateway: state handle + handle to the server's background task. pub struct RunningSeller { pub state: Arc, - /// The seller's note -- **polymorphic**: `LocalNote` (mock path) OR `RealNote` (a real chain, - /// one SDK key for signing+handover). The gateway encrypts the endpoint `note.encrypt_to(buyer_pubkey)` -- on + /// The seller's note — **polymorphic** (D10): `LocalNote` (mock path) OR `RealNote` (a real chain, + /// one SDK key for signing+handover). The gateway encrypts the endpoint `note.encrypt_to(buyer_pubkey)` — on /// the real path `buyer_pubkey` is reconstructed by the seller from on-chain ed25519 (F1). pub note: Arc, pub server_task: tokio::task::JoinHandle<()>, /// The socket address actually bound before the server task was spawned. pub listen_addr: SocketAddr, - /// Fingerprint of the gateway's self-signed TLS certificate -- goes into the handover. + /// Fingerprint of the gateway's self-signed TLS certificate (§3.1.3) — goes into the handover. pub tls_fingerprint: String, } @@ -112,6 +115,69 @@ struct SellerMatchWatchCursor { opened_at_unix: Option, #[serde(default)] fill: Option, + /// A write-ahead marker for `postSellOffer`. It is retained until an + /// authoritative exact-TC reconciliation proves what the write did. This + /// must live with the durable watch cursor, rather than in the in-memory + /// chain adapter, because a service restart is the failure case it protects. + #[serde(default)] + publication: Option, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub(crate) struct SellerOfferSubmission { + version: u32, + #[serde(with = "dexdo_core::address::serde_self_dapp")] + token_contract: TokenContract, + /// Lower bound for the owner-note event scan. Unlike the old backend-local + /// timestamp, this survives a process restart. + #[serde(default)] + event_since_unix: Option, + submitted_at_unix: u64, + /// A one-shot permit created only by the explicit audited recovery command. + /// Normal service/controller startup deliberately never consumes it. + #[serde(default)] + manual_recovery: Option, +} + +impl SellerOfferSubmission { + fn event_since_unix(&self) -> u64 { + self.event_since_unix.unwrap_or(self.submitted_at_unix) + } +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct SellerOfferManualRecoveryPermit { + version: u32, + audit_path: String, + audit_sha256: String, + authorized_at_unix: u64, + #[serde(default)] + consumed_at_unix: Option, +} + +#[derive(serde::Serialize)] +struct SellerOfferManualRecoveryAudit<'a> { + version: u32, + #[serde(with = "dexdo_core::address::serde_self_dapp")] + token_contract: &'a TokenContract, + marker_sha256: String, + evidence_cursor_unix: u64, + evidence_outcome: &'static str, + authorized_at_unix: u64, +} + +#[derive(serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct SellerOfferManualRecoveryAuditRead { + version: u32, + #[serde(with = "dexdo_core::address::serde_self_dapp")] + token_contract: TokenContract, + marker_sha256: String, + evidence_cursor_unix: u64, + evidence_outcome: String, + authorized_at_unix: u64, } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] @@ -158,6 +224,7 @@ impl SellerMatchWatchCursor { last_polled_unix: None, opened_at_unix: None, fill: None, + publication: None, }) } @@ -278,21 +345,382 @@ impl SellerMatchWatchCursor { })?; } } - let tmp = path.with_extension(format!("json.tmp.{}", std::process::id())); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|error| anyhow!("seller watch cursor clock before epoch: {error}"))? + .as_nanos(); + let tmp = path.with_extension(format!("json.tmp.{}.{}", std::process::id(), nanos)); let bytes = serde_json::to_vec_pretty(self)?; - std::fs::write(&tmp, bytes).map_err(|e| { - anyhow::anyhow!("write seller watch cursor temp {}: {e}", tmp.display()) - })?; + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&tmp) + .map_err(|e| { + anyhow::anyhow!("create seller watch cursor temp {}: {e}", tmp.display()) + })?; + file.write_all(&bytes) + .and_then(|()| file.sync_all()) + .map_err(|e| { + anyhow::anyhow!("write seller watch cursor temp {}: {e}", tmp.display()) + })?; std::fs::rename(&tmp, path).map_err(|e| { anyhow::anyhow!( "commit seller watch cursor {} from temp {}: {e}", path.display(), tmp.display() ) - }) + })?; + if let Some(parent) = path.parent() { + if !parent.as_os_str().is_empty() { + sync_seller_cursor_parent(parent)?; + } + } + Ok(()) } } +/// A rename is not crash-durable on Unix until its containing directory has +/// been synced. Windows' rename API is already write-through at this layer; +/// `File::sync_all` on a directory is not supported there. +#[cfg(unix)] +fn sync_seller_cursor_parent(parent: &Path) -> Result<()> { + std::fs::File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(|error| anyhow!("sync seller watch cursor dir {}: {error}", parent.display())) +} + +#[cfg(not(unix))] +fn sync_seller_cursor_parent(_parent: &Path) -> Result<()> { + Ok(()) +} + +/// Serializes durable publication-marker mutation across independently started +/// CLI and service processes. The marker itself is the long-lived fence; this +/// short-lived lock only closes the read-none/write-marker race. +struct SellerOfferSubmissionLock { + path: PathBuf, + _file: std::fs::File, +} + +impl Drop for SellerOfferSubmissionLock { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + if let Some(parent) = self.path.parent() { + let _ = sync_seller_cursor_parent(parent); + } + } +} + +fn lock_offer_submission(cursor_path: &Path) -> Result { + let parent = cursor_path.parent().ok_or_else(|| { + anyhow!( + "seller publication marker {} has no parent directory", + cursor_path.display() + ) + })?; + std::fs::create_dir_all(parent).map_err(|error| { + anyhow!( + "create seller publication marker directory {}: {error}", + parent.display() + ) + })?; + let path = cursor_path.with_extension("publication.lock"); + let file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + .map_err(|error| { + anyhow!( + "publication_unconfirmed token_contract cursor={}: durable publication lock {} is held or unreadable ({error}); no automatic postSellOffer retry is permitted", + cursor_path.display(), + path.display() + ) + })?; + file.sync_all() + .map_err(|error| anyhow!("sync seller publication lock {}: {error}", path.display()))?; + sync_seller_cursor_parent(parent)?; + Ok(SellerOfferSubmissionLock { path, _file: file }) +} + +pub(crate) fn pending_offer_submission( + cursor_path: &Path, + token_contract: &TokenContract, +) -> Result> { + let cursor = SellerMatchWatchCursor::load_or_new(cursor_path, token_contract)?; + if let Some(marker) = cursor.publication.as_ref() { + if marker.version != SELLER_OFFER_SUBMISSION_VERSION { + bail!( + "seller publication marker {} has version {}; expected {}", + cursor_path.display(), + marker.version, + SELLER_OFFER_SUBMISSION_VERSION + ); + } + if !marker.token_contract.eq_ignore_ascii_case(token_contract) { + bail!( + "seller publication marker {} is for TokenContract {}, not {}", + cursor_path.display(), + display_token_contract(&marker.token_contract), + display_token_contract(token_contract) + ); + } + } + Ok(cursor.publication) +} + +pub(crate) fn persist_offer_submission( + cursor_path: &Path, + token_contract: &TokenContract, +) -> Result { + let _lock = lock_offer_submission(cursor_path)?; + let mut cursor = SellerMatchWatchCursor::load_or_new(cursor_path, token_contract)?; + if cursor.publication.is_some() { + bail!( + "seller publication marker {} already exists for {}; reconcile it before another postSellOffer", + cursor_path.display(), + display_token_contract(token_contract) + ); + } + let submitted_at_unix = now_unix()?; + let marker = SellerOfferSubmission { + version: SELLER_OFFER_SUBMISSION_VERSION, + token_contract: token_contract.clone(), + // This margin is a durable event cursor, not an in-memory post clock. + event_since_unix: Some( + submitted_at_unix.saturating_sub(dexdo_core::params::SELLER_OFFER_EVENT_LOOKBACK_SECS), + ), + submitted_at_unix, + manual_recovery: None, + }; + cursor.publication = Some(marker.clone()); + cursor.save(cursor_path)?; + Ok(marker) +} + +pub(crate) fn clear_offer_submission( + cursor_path: &Path, + token_contract: &TokenContract, +) -> Result<()> { + let _lock = lock_offer_submission(cursor_path)?; + let mut cursor = SellerMatchWatchCursor::load_or_new(cursor_path, token_contract)?; + cursor.publication = None; + cursor.save(cursor_path) +} + +fn sha256_hex(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +fn write_immutable_manual_recovery_audit( + cursor_path: &Path, + token_contract: &TokenContract, + marker: &SellerOfferSubmission, + authorized_at_unix: u64, +) -> Result { + let marker_sha256 = sha256_hex(&serde_json::to_vec(marker)?); + let audit = SellerOfferManualRecoveryAudit { + version: 1, + token_contract, + marker_sha256, + evidence_cursor_unix: marker.event_since_unix(), + evidence_outcome: "exact_negative", + authorized_at_unix, + }; + let audit_bytes = serde_json::to_vec_pretty(&audit)?; + let audit_sha256 = sha256_hex(&audit_bytes); + let parent = cursor_path.parent().ok_or_else(|| { + anyhow!( + "seller publication marker {} has no parent directory", + cursor_path.display() + ) + })?; + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|error| anyhow!("manual recovery clock before epoch: {error}"))? + .as_nanos(); + let audit_path = parent.join(format!( + ".{}.publication-recovery.{}.{}.json", + cursor_path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("seller-watch"), + authorized_at_unix, + nanos + )); + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&audit_path) + .map_err(|error| { + anyhow!( + "create immutable manual recovery audit {}: {error}", + audit_path.display() + ) + })?; + file.write_all(&audit_bytes) + .and_then(|()| file.sync_all()) + .map_err(|error| { + anyhow!( + "write immutable manual recovery audit {}: {error}", + audit_path.display() + ) + })?; + sync_seller_cursor_parent(parent)?; + Ok(SellerOfferManualRecoveryPermit { + version: 1, + audit_path: audit_path.display().to_string(), + audit_sha256, + authorized_at_unix, + consumed_at_unix: None, + }) +} + +/// Explicitly authorize one fresh seller start after an old marked publication +/// has been proven absent. This is deliberately separate from ordinary +/// startup: a controller/timer cannot obtain this permit by merely restarting. +pub(crate) async fn authorize_exact_negative_manual_recovery( + chain: &dyn ChainBackend, + cfg: &SellerConfig, + expected_owner: &str, + cursor_path: &Path, + operator_token_contract: &str, +) -> Result<()> { + let operator_token_contract_normalized = normalize_wallet_address(operator_token_contract) + .map_err(|error| { + anyhow!("manual publication recovery TokenContract is invalid: {error}") + })?; + let selected_token_contract_normalized = normalize_wallet_address(&cfg.token_contract) + .map_err(|error| anyhow!("selected TokenContract is invalid: {error}"))?; + if operator_token_contract_normalized != selected_token_contract_normalized { + bail!("manual publication recovery contract does not match the selected TokenContract"); + } + let _lock = lock_offer_submission(cursor_path)?; + let mut cursor = SellerMatchWatchCursor::load_or_new(cursor_path, &cfg.token_contract)?; + let marker = cursor.publication.as_ref().ok_or_else(|| { + publication_unconfirmed( + &cfg.token_contract, + "manual recovery requires an existing retained publication marker", + ) + })?; + if marker.manual_recovery.is_some() { + bail!( + "manual publication recovery is already authorized for {}; a controller cannot consume it and a second authorization is refused", + display_token_contract(&cfg.token_contract) + ); + } + match reconcile_pending_offer_submission(chain, cfg, Some(expected_owner), marker).await? { + PendingPublicationResolution::Negative => {} + PendingPublicationResolution::Resting { order_id } => bail!( + "manual publication recovery refused for {}: exact raw SELL is resting as order {order_id}", + display_token_contract(&cfg.token_contract) + ), + PendingPublicationResolution::Funded => bail!( + "manual publication recovery refused for {}: exact match is funded", + display_token_contract(&cfg.token_contract) + ), + } + let authorized_at_unix = now_unix()?; + // The audit is committed first. A crash before the cursor permit is saved + // leaves an orphan immutable record but never enables a repost. + let permit = write_immutable_manual_recovery_audit( + cursor_path, + &cfg.token_contract, + marker, + authorized_at_unix, + )?; + cursor + .publication + .as_mut() + .expect("marker was checked above") + .manual_recovery = Some(permit); + cursor.save(cursor_path) +} + +/// Consume a durable audited permit before the single explicit post path. +/// The consumed bit is persisted before any chain write, so a crash cannot +/// turn a service restart into another manual recovery attempt. +pub(crate) fn consume_exact_negative_manual_recovery( + cursor_path: &Path, + token_contract: &TokenContract, +) -> Result<()> { + let _lock = lock_offer_submission(cursor_path)?; + let mut cursor = SellerMatchWatchCursor::load_or_new(cursor_path, token_contract)?; + let marker = cursor.publication.as_mut().ok_or_else(|| { + publication_unconfirmed( + token_contract, + "manual recovery permit disappeared before the explicit post", + ) + })?; + let expected_cursor = marker.event_since_unix(); + let expected_marker_sha256 = sha256_hex(&serde_json::to_vec(&SellerOfferSubmission { + version: marker.version, + token_contract: marker.token_contract.clone(), + event_since_unix: marker.event_since_unix, + submitted_at_unix: marker.submitted_at_unix, + manual_recovery: None, + })?); + let permit = marker.manual_recovery.as_mut().ok_or_else(|| { + publication_unconfirmed( + token_contract, + "manual recovery has no durable audited permit", + ) + })?; + if permit.consumed_at_unix.is_some() { + bail!( + "manual publication recovery permit for {} was already consumed; automatic restart remains refused", + display_token_contract(token_contract) + ); + } + let audit_bytes = std::fs::read(&permit.audit_path).map_err(|error| { + publication_unconfirmed( + token_contract, + format!( + "audited manual recovery record {} is missing or unreadable: {error}", + permit.audit_path + ), + ) + })?; + if sha256_hex(&audit_bytes) != permit.audit_sha256 { + return Err(publication_unconfirmed( + token_contract, + format!( + "audited manual recovery record {} SHA-256 does not match its durable permit", + permit.audit_path + ), + )); + } + let audit: SellerOfferManualRecoveryAuditRead = + serde_json::from_slice(&audit_bytes).map_err(|error| { + publication_unconfirmed( + token_contract, + format!( + "audited manual recovery record {} is invalid: {error}", + permit.audit_path + ), + ) + })?; + if audit.version != 1 + || !audit.token_contract.eq_ignore_ascii_case(token_contract) + || audit.marker_sha256 != expected_marker_sha256 + || audit.evidence_cursor_unix != expected_cursor + || audit.evidence_outcome != "exact_negative" + || audit.authorized_at_unix != permit.authorized_at_unix + { + return Err(publication_unconfirmed( + token_contract, + format!( + "audited manual recovery record {} is not bound to this retained publication marker", + permit.audit_path + ), + )); + } + permit.consumed_at_unix = Some(now_unix()?); + cursor.save(cursor_path) +} + pub fn read_seller_fill_lineage( cursor_path: &Path, token_contract: &TokenContract, @@ -347,7 +775,7 @@ fn now_unix() -> Result { .as_secs()) } -/// Bring up the seller's gRPC gateway (headless) **over TLS**: a self-signed certificate +/// Bring up the seller's gRPC gateway (headless) **over TLS** (§3.1.3): a self-signed certificate /// is generated at startup, its fingerprint is returned for recording in the handover. Returns /// handles for orchestrating the stream. pub async fn start_gateway(addr: SocketAddr) -> Result { @@ -355,17 +783,17 @@ pub async fn start_gateway(addr: SocketAddr) -> Result { } /// Like [`start_gateway`], but with an upstream choice (mock model or real OpenAI-compatible, -/// ). The mock path (`UpstreamConfig::Mock`) is identical to. +/// Directive 3). The mock path (`UpstreamConfig::Mock`) is identical to Directive 1. pub async fn start_gateway_with( addr: SocketAddr, upstream: UpstreamConfig, ) -> Result { - // The ephemeral note is a mock fixture; the production path is `start_gateway_with_note`. + // The ephemeral note is a mock fixture (Directive 7); the production path is `start_gateway_with_note`. start_gateway_with_note(addr, upstream, Arc::new(LocalNote::generate())).await } -/// Like [`start_gateway_with`], but with a **loaded persistent** seller note: -/// the identity (from `--note-key`/wallet) is reused across runs -- its offer/deals are +/// Like [`start_gateway_with`], but with a **loaded persistent** seller note (Directive 7): +/// the identity (from `--note-key`/wallet) is reused across runs — its offer/deals are /// visible in the next run. `start_gateway_with` substitutes an ephemeral `generate()` here. pub async fn start_gateway_with_note( addr: SocketAddr, @@ -419,7 +847,7 @@ async fn start_gateway_with_note_and_capacity_dir( deals_dir: Option, gw_tls: GatewayTls, ) -> Result { - // this process is about to serve buyers, and a buyer that hangs up mid-stream must give + // #1571: this process is about to serve buyers, and a buyer that hangs up mid-stream must give // this gateway an `EPIPE` to handle rather than a signal that kills it. `main` restores the // default SIGPIPE disposition for one-shot printers; a gateway needs it ignored. Do not delete // this as a duplicate of the entry policy -- it is the opposite decision, for the opposite kind @@ -432,7 +860,7 @@ async fn start_gateway_with_note_and_capacity_dir( let service = GatewayService::new(state.clone()).into_server(); // Both rustls providers (ring/aws-lc-rs) are present in the tree; pin the process - // default explicitly (ring) -- otherwise rustls panics, unable to pick on its own. Idempotent. + // default explicitly (ring) — otherwise rustls panics, unable to pick on its own. Idempotent. tls::ensure_crypto_provider(); let tls_fingerprint = gw_tls.fingerprint.clone(); @@ -467,7 +895,7 @@ async fn start_gateway_with_note_and_capacity_dir( }) } -/// Post a sell offer from the note into the book. Done before the +/// Post a sell offer from the note into the book (§10.3 step 2, §2.1). Done before the /// buyer places a buy order. pub async fn post_offer( seller: &RunningSeller, @@ -719,11 +1147,176 @@ pub async fn prepare_seller_offer( } } -/// Open the stream for a match: -/// 1. reads the match (the buyer's pubkey is recorded in the contract); -/// 2. encrypts the endpoint to the buyer's pubkey and `open_stream` (probe freeze + -/// exact `2P` seller bond + writing the enc-endpoint into the endpoints file); -/// 3. registers the buyer's pubkey and the fake-token budget in the gateway for authorization. +/// The terminal fact for a previously submitted seller publication marker. +/// +/// `Negative` does not mean "post again now". It means every required read +/// completed and proved that the earlier write made no offer. The current +/// invocation still returns `publication_unconfirmed`; a later explicit +/// seller start may make the next write. This keeps a transport failure from +/// becoming an invisible repost loop. +pub(crate) enum PendingPublicationResolution { + Resting { order_id: u128 }, + Funded, + Negative, +} + +fn publication_unconfirmed( + token_contract: &TokenContract, + detail: impl std::fmt::Display, +) -> anyhow::Error { + anyhow!( + "publication_unconfirmed token_contract={}: {detail}; no automatic postSellOffer retry is permitted", + display_token_contract(token_contract) + ) +} + +/// Reconcile a persisted write-ahead marker solely from exact, read-only facts. +/// Every source is required for the negative verdict: an unavailable raw book, +/// match read, offer latch, or event/outcome read leaves the marker in place. +pub(crate) async fn reconcile_pending_offer_submission( + chain: &dyn ChainBackend, + cfg: &SellerConfig, + expected_owner: Option<&str>, + marker: &SellerOfferSubmission, +) -> Result { + match inspect_seller_offer(chain, cfg, expected_owner).await { + Ok(SellerOfferInspection::Funded) => return Ok(PendingPublicationResolution::Funded), + Ok(SellerOfferInspection::Resting { order_id }) => { + return Ok(PendingPublicationResolution::Resting { order_id }); + } + Ok(SellerOfferInspection::Vacant) => {} + Err(error) => return Err(publication_unconfirmed(&cfg.token_contract, error)), + } + + let latch = chain + .seller_offer_latch(&cfg.token_contract) + .await + .map_err(|error| { + publication_unconfirmed( + &cfg.token_contract, + format!("exact TokenContract offer latch is unreadable: {error}"), + ) + })? + .ok_or_else(|| { + publication_unconfirmed( + &cfg.token_contract, + "exact TokenContract offer latch is unavailable", + ) + })?; + + let outcome = chain + .seller_offer_outcome_since(&cfg.token_contract, marker.event_since_unix()) + .await + .map_err(|error| { + publication_unconfirmed( + &cfg.token_contract, + format!( + "marker-bounded event/outcome reconciliation since {} is unreadable: {error}", + marker.event_since_unix() + ), + ) + })?; + match outcome { + Some(SellOfferOutcome::Matched) => return Ok(PendingPublicationResolution::Funded), + Some(SellOfferOutcome::Rested { order_id }) => { + return Ok(PendingPublicationResolution::Resting { order_id }); + } + None => {} + } + if latch.offer_posted { + return Err(publication_unconfirmed( + &cfg.token_contract, + "TokenContract getOffer().offerPosted=true while no exact raw SELL row is visible", + )); + } + Ok(PendingPublicationResolution::Negative) +} + +/// Persistent counterpart of [`prepare_seller_offer`] for production seller +/// runs. It records a marker before the only chain write and makes every +/// subsequent start reconcile that marker before it can submit again. +pub async fn prepare_seller_offer_with_persisted_submission( + note: &dyn Note, + chain: &dyn ChainBackend, + cfg: &SellerConfig, + expected_owner: Option<&str>, + cursor_path: &Path, +) -> Result { + if let Some(marker) = pending_offer_submission(cursor_path, &cfg.token_contract)? { + match reconcile_pending_offer_submission(chain, cfg, expected_owner, &marker).await { + Ok(PendingPublicationResolution::Resting { order_id }) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + return Ok(SellerOfferStartup::ResumedResting { order_id }); + } + Ok(PendingPublicationResolution::Funded) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + return Ok(SellerOfferStartup::ResumedFunded); + } + Ok(PendingPublicationResolution::Negative) => { + return Err(publication_unconfirmed( + &cfg.token_contract, + format!( + "submission marker from {} has an exact negative proof (no match, raw SELL, offer latch, or event outcome); marker is retained so an automatic restart cannot retry — use an audited manual recovery operation", + marker.submitted_at_unix + ), + )); + } + Err(error) => return Err(error), + } + } + + match inspect_seller_offer(chain, cfg, expected_owner).await? { + SellerOfferInspection::Funded => Ok(SellerOfferStartup::ResumedFunded), + SellerOfferInspection::Resting { order_id } => { + Ok(SellerOfferStartup::ResumedResting { order_id }) + } + SellerOfferInspection::Vacant => { + chain + .assert_token_contract_fresh(&cfg.token_contract) + .await?; + // Write-ahead persistence is deliberately before the write. A + // crash in the narrow interval thereafter is conservative: the + // next process reconciles instead of guessing the write did not run. + let marker = persist_offer_submission(cursor_path, &cfg.token_contract)?; + let _submit_error = post_offer_with_note(note, chain, cfg).await.err(); + match reconcile_pending_offer_submission(chain, cfg, expected_owner, &marker).await { + Ok(PendingPublicationResolution::Resting { order_id }) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + Ok(SellerOfferStartup::Posted { + outcome: Some(SellOfferOutcome::Rested { order_id }), + }) + } + Ok(PendingPublicationResolution::Funded) => { + clear_offer_submission(cursor_path, &cfg.token_contract)?; + Ok(SellerOfferStartup::Posted { + outcome: Some(SellOfferOutcome::Matched), + }) + } + Ok(PendingPublicationResolution::Negative) => { + Err(publication_unconfirmed( + &cfg.token_contract, + "postSellOffer has an exact negative proof; marker is retained so an automatic restart cannot retry — use an audited manual recovery operation", + )) + } + Err(error) => { + if let Some(submit_error) = _submit_error { + return Err(publication_unconfirmed( + &cfg.token_contract, + format!("postSellOffer returned {submit_error}; {error}"), + )); + } + Err(error) + } + } + } + } +} + +/// Open the stream for a match (§10.3 step 3): +/// 1. reads the match (the buyer's pubkey is recorded in the contract); +/// 2. encrypts the endpoint to the buyer's pubkey and `open_stream` (probe freeze + +/// exact `2P` seller bond + writing the enc-endpoint into the endpoints file); +/// 3. registers the buyer's pubkey and the fake-token budget in the gateway for authorization. pub async fn serve_match( seller: &RunningSeller, chain: &dyn ChainBackend, @@ -783,7 +1376,7 @@ pub async fn provision_match( display_token_contract(&cfg.token_contract) ); } - // the handover {gateway endpoint, TLS fingerprint} is encrypted to the buyer's pubkey. + // §3.1/§3.1.3: the handover {gateway endpoint, TLS fingerprint} is encrypted to the buyer's pubkey. // The endpoint points at the GATEWAY over TLS (R15); the buyer pins the fingerprint on connect. let handover = Handover { endpoint: format!("https://{}", cfg.gateway_advertise), @@ -796,11 +1389,11 @@ pub async fn provision_match( &handover.to_deal_bytes(&cfg.token_contract), ); - // the gateway must authorize the matched buyer BEFORE that buyer can connect. + // §3.1.1: the gateway must authorize the matched buyer BEFORE that buyer can connect. // Register buyer+budget BEFORE writing the handover on-chain: the buyer learns the endpoint only // after reading the on-chain ciphertext (written by `open_stream`), so register-before-open rules out a race. Otherwise on a // real (slow) chain the buyer manages to knock in the window between open_stream and register_stream - // -> the gateway still has no pubkey -> `challenge-response failed` (the mock timing did not expose this). + // → the gateway still has no pubkey → `challenge-response failed` (the mock timing did not expose this). let (state, deal) = read_coherent_deal_capacity(chain, &cfg.token_contract).await?; if cfg.subscription != deal.is_subscription() { let expected = if cfg.subscription { @@ -964,7 +1557,7 @@ pub async fn poll_match_and_maybe_open( } /// Wait for one authoritative match without beginning the on-chain handover write. - +/// /// Keeping this phase read-only lets the resting-offer supervisor select shutdown/health safely. Once a /// match is observed, [`serve_watched_match`] runs the existing handover path to completion outside that /// cancellable select. @@ -1033,7 +1626,8 @@ mod tests { use super::*; use dexdo_core::{ validate_seller_resume_state, ChainError, DealBuyerBond, DealChainSnapshot, DealChainState, - DealSellerBond, LocalNote, NotePubkey, OfferListing, SellOffer, Settlement, StreamSnapshot, + DealOfferLatch, DealSellerBond, LocalNote, NotePubkey, OfferListing, SellOffer, + SellOfferOutcome, Settlement, StreamSnapshot, }; use dexdo_proto::{CanonRequest, ChallengeRequest, GatewayClient, StreamRequest}; use proptest::prelude::*; @@ -1406,7 +2000,10 @@ mod tests { } struct StartupBackend { - raw: RawStartupRead, + raw: Mutex, + outcome: Option, + outcome_failure: Option, + offer_latch: Option, startup_match: Option, watcher_match: Option, post_calls: AtomicU64, @@ -1423,7 +2020,10 @@ mod tests { impl StartupBackend { fn new(raw: RawStartupRead, startup_match: Option, watcher_match: Match) -> Self { Self { - raw, + raw: Mutex::new(raw), + outcome: Some(SellOfferOutcome::Rested { order_id: 835 }), + outcome_failure: None, + offer_latch: None, startup_match, watcher_match: Some(watcher_match), post_calls: AtomicU64::new(0), @@ -1440,7 +2040,10 @@ mod tests { fn without_match(raw: RawStartupRead) -> Self { Self { - raw, + raw: Mutex::new(raw), + outcome: Some(SellOfferOutcome::Rested { order_id: 835 }), + outcome_failure: None, + offer_latch: None, startup_match: None, watcher_match: None, post_calls: AtomicU64::new(0), @@ -1469,6 +2072,21 @@ mod tests { self.resume_facts = Some((state, price_per_tick)); self } + + fn with_offer_outcome(mut self, outcome: Option) -> Self { + self.outcome = outcome; + self + } + + fn with_offer_outcome_failure(mut self, failure: impl Into) -> Self { + self.outcome_failure = Some(failure.into()); + self + } + + fn with_offer_latch(mut self, offer_posted: bool) -> Self { + self.offer_latch = Some(offer_posted); + self + } } #[async_trait::async_trait] @@ -1482,7 +2100,7 @@ mod tests { _: &TokenContract, ) -> Result, ChainError> { self.raw_reads.fetch_add(1, Ordering::Relaxed); - match &self.raw { + match &*self.raw.lock().unwrap() { RawStartupRead::Orders(orders) => Ok(orders.clone()), RawStartupRead::ChainFailure => { Err(ChainError::Chain("raw book getter failed".to_string())) @@ -1508,7 +2126,30 @@ mod tests { &self, _: &TokenContract, ) -> Result, ChainError> { - Ok(Some(SellOfferOutcome::Rested { order_id: 835 })) + if let Some(failure) = &self.outcome_failure { + return Err(ChainError::Transport(failure.clone())); + } + Ok(self.outcome.clone()) + } + + async fn seller_offer_outcome_since( + &self, + _: &TokenContract, + _: u64, + ) -> Result, ChainError> { + if let Some(failure) = &self.outcome_failure { + return Err(ChainError::Transport(failure.clone())); + } + Ok(self.outcome.clone()) + } + + async fn seller_offer_latch( + &self, + _: &TokenContract, + ) -> Result, ChainError> { + Ok(self + .offer_latch + .map(|offer_posted| DealOfferLatch { offer_posted })) } async fn read_openable_match_now( @@ -1690,8 +2331,8 @@ mod tests { } } - /// the directory is returned with the path and must be held for as long as the cursor is - /// read or written -- the previous `-` directory was never removed. + /// #822: the directory is returned with the path and must be held for as long as the cursor is + /// read or written — the previous `-` directory was never removed. fn temp_cursor_path(name: &str) -> (tempfile::TempDir, PathBuf) { let dir = tempfile::Builder::new() .prefix("dexdo-seller-watch-test") @@ -1734,38 +2375,394 @@ mod tests { } } - /// Issues ** /** at a money-path consumer, where the equality DECIDES something. + #[tokio::test] + async fn manual_publication_recovery_requires_new_exact_negative_proof_and_writes_audit() { + let tc = chain_address('a'); + let owner = chain_address('b'); + let cfg = test_cfg(&tc); + let backend = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome(None); + let (dir, cursor) = temp_cursor_path("manual-publication-negative"); + persist_offer_submission(&cursor, &tc).unwrap(); + + // The explicit operator command may use canonical display form while the + // persisted market/config retains the legacy contract-parameter form. + let canonical_tc = dexdo_core::address::display_self_dapp(&tc); + authorize_exact_negative_manual_recovery(&backend, &cfg, &owner, &cursor, &canonical_tc) + .await + .expect("equivalent canonical address may authorize exact-negative recovery"); + let marker = pending_offer_submission(&cursor, &tc) + .unwrap() + .expect("marker stays until the one-shot post is reconciled"); + let permit = marker.manual_recovery.expect("durable manual permit"); + let audit = std::fs::read_to_string(&permit.audit_path).expect("immutable audit exists"); + assert!(audit.contains("exact_negative"), "{audit}"); + assert!(audit.contains("marker_sha256"), "{audit}"); + assert!(audit.contains("evidence_cursor_unix"), "{audit}"); + assert!(permit.consumed_at_unix.is_none()); + assert!(permit + .audit_path + .starts_with(&dir.path().display().to_string())); + + consume_exact_negative_manual_recovery(&cursor, &tc).expect("consume exactly once"); + let second = consume_exact_negative_manual_recovery(&cursor, &tc) + .expect_err("crash/restart cannot consume the same permit twice"); + assert!( + second.to_string().contains("already consumed"), + "{second:#}" + ); + } + + #[tokio::test] + async fn manual_publication_recovery_refuses_positive_or_unreadable_evidence() { + let tc = chain_address('a'); + let owner = chain_address('b'); + let cfg = test_cfg(&tc); + let resting = raw_sell( + 77, + &owner, + Some(&tc), + u128::from(cfg.price_per_tick), + u128::from(cfg.max_ticks), + ); + let positive = StartupBackend::without_match(RawStartupRead::Orders(vec![resting])); + let (_dir, cursor) = temp_cursor_path("manual-publication-positive"); + persist_offer_submission(&cursor, &tc).unwrap(); + let error = authorize_exact_negative_manual_recovery(&positive, &cfg, &owner, &cursor, &tc) + .await + .expect_err("resting exact SELL must refuse manual recovery"); + assert!(error.to_string().contains("resting"), "{error:#}"); + assert!(pending_offer_submission(&cursor, &tc) + .unwrap() + .unwrap() + .manual_recovery + .is_none()); - /// The helpers that drop the dapp id -- `to_chain_param`, `normalize_wallet_address`, - /// `parse_chain_address` -- are all documented to yield the workchain form, so asserting that + let unreadable = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome_failure("event API timeout"); + let (_dir, cursor) = temp_cursor_path("manual-publication-unreadable"); + persist_offer_submission(&cursor, &tc).unwrap(); + let error = + authorize_exact_negative_manual_recovery(&unreadable, &cfg, &owner, &cursor, &tc) + .await + .expect_err("unreadable evidence must remain fail closed"); + assert!( + error.to_string().contains("publication_unconfirmed"), + "{error:#}" + ); + assert!(pending_offer_submission(&cursor, &tc) + .unwrap() + .unwrap() + .manual_recovery + .is_none()); + } + + #[tokio::test] + async fn manual_publication_recovery_lock_excludes_a_second_operator() { + let tc = chain_address('a'); + let owner = chain_address('b'); + let cfg = test_cfg(&tc); + let backend = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome(None); + let (_dir, cursor) = temp_cursor_path("manual-publication-lock"); + persist_offer_submission(&cursor, &tc).unwrap(); + let held = lock_offer_submission(&cursor).expect("hold first operator lock"); + let error = authorize_exact_negative_manual_recovery(&backend, &cfg, &owner, &cursor, &tc) + .await + .expect_err("second operator must not obtain a permit concurrently"); + assert!( + error.to_string().contains("durable publication lock"), + "{error:#}" + ); + drop(held); + } + + #[tokio::test] + async fn manual_publication_recovery_refuses_missing_or_tampered_audit_before_consumption() { + let tc = chain_address('a'); + let owner = chain_address('b'); + let cfg = test_cfg(&tc); + let backend = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome(None); + + let (_dir, cursor) = temp_cursor_path("manual-publication-missing-audit"); + persist_offer_submission(&cursor, &tc).unwrap(); + authorize_exact_negative_manual_recovery(&backend, &cfg, &owner, &cursor, &tc) + .await + .unwrap(); + let audit_path = pending_offer_submission(&cursor, &tc) + .unwrap() + .unwrap() + .manual_recovery + .unwrap() + .audit_path; + std::fs::remove_file(&audit_path).unwrap(); + let error = consume_exact_negative_manual_recovery(&cursor, &tc) + .expect_err("a deleted audit must not permit a post"); + assert!( + error.to_string().contains("missing or unreadable"), + "{error:#}" + ); + assert!(pending_offer_submission(&cursor, &tc) + .unwrap() + .unwrap() + .manual_recovery + .unwrap() + .consumed_at_unix + .is_none()); + + let (_dir, cursor) = temp_cursor_path("manual-publication-tampered-audit"); + persist_offer_submission(&cursor, &tc).unwrap(); + authorize_exact_negative_manual_recovery(&backend, &cfg, &owner, &cursor, &tc) + .await + .unwrap(); + let audit_path = pending_offer_submission(&cursor, &tc) + .unwrap() + .unwrap() + .manual_recovery + .unwrap() + .audit_path; + std::fs::write(&audit_path, b"tampered").unwrap(); + let error = consume_exact_negative_manual_recovery(&cursor, &tc) + .expect_err("a modified audit must not permit a post"); + assert!(error.to_string().contains("SHA-256"), "{error:#}"); + assert!(pending_offer_submission(&cursor, &tc) + .unwrap() + .unwrap() + .manual_recovery + .unwrap() + .consumed_at_unix + .is_none()); + } + + #[tokio::test] + async fn publication_marker_reconciles_late_exact_resting_sell_without_repost() { + let tc = chain_address('a'); + let owner = chain_address('b'); + let cfg = test_cfg(&tc); + let late_order = raw_sell( + 44, + &owner, + Some(&tc), + u128::from(cfg.price_per_tick), + u128::from(cfg.max_ticks), + ); + let backend = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome_failure("book/event read temporarily unavailable"); + let note = LocalNote::generate(); + let (_dir, cursor) = temp_cursor_path("publication-late-resting"); + + // The write succeeds but the post-write outcome endpoint times out. The + // write-ahead marker survives and the seller is explicitly unavailable. + let first = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .expect_err("unavailable exact outcome must remain unconfirmed"); + assert!(first.to_string().contains("publication_unconfirmed")); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 1); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_some()); + + // The delayed exact row becomes visible after the process stopped. + *backend.raw.lock().unwrap() = RawStartupRead::Orders(vec![late_order]); + + // A process restart adopts the exact row and never emits a duplicate + // postSellOffer for the same TokenContract. + let second = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .unwrap(); + assert_eq!(second, SellerOfferStartup::ResumedResting { order_id: 44 }); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn publication_marker_stays_fail_closed_when_all_authoritative_reads_fail() { + let tc = chain_address('c'); + let owner = chain_address('d'); + let cfg = test_cfg(&tc); + let backend = StartupBackend::without_match(RawStartupRead::ChainFailure) + .with_offer_latch(false) + .with_offer_outcome_failure("events unavailable"); + let note = LocalNote::generate(); + let (_dir, cursor) = temp_cursor_path("publication-all-reads-fail"); + persist_offer_submission(&cursor, &tc).unwrap(); + + let error = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .expect_err("unreadable exact book must not permit a new post"); + assert!(error.to_string().contains("publication_unconfirmed")); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 0); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_some()); + } + + #[tokio::test] + async fn publication_marker_negative_proof_is_retained_and_blocks_automatic_restarts() { + let tc = chain_address('e'); + let owner = chain_address('f'); + let cfg = test_cfg(&tc); + let backend = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome(None); + let note = LocalNote::generate(); + let (_dir, cursor) = temp_cursor_path("publication-negative-proof"); + + let first = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .expect_err("a negative proof is not an in-process repost permit"); + assert!(first.to_string().contains("publication_unconfirmed")); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 1); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_some()); + + // A process supervisor cannot masquerade as a fresh manual request: + // the durable negative marker remains and no second post is emitted. + let restart = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .expect_err("automatic restart must stay fenced by the negative marker"); + assert!(restart.to_string().contains("publication_unconfirmed")); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 1); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_some()); + } + + #[test] + fn publication_marker_lock_excludes_a_second_cli_or_service_start() { + let tc = chain_address('d'); + let (_dir, cursor) = temp_cursor_path("publication-exclusive-lock"); + let guard = lock_offer_submission(&cursor).expect("acquire first durable marker lock"); + let blocked = persist_offer_submission(&cursor, &tc) + .expect_err("second start must not race marker creation"); + assert!(blocked.to_string().contains("publication_unconfirmed")); + drop(guard); + persist_offer_submission(&cursor, &tc) + .expect("marker can be committed after the first exclusive lock is released"); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_some()); + } + + #[tokio::test] + async fn publication_marker_adopts_immediate_match_after_accepted_write() { + let tc = chain_address('1'); + let owner = chain_address('2'); + let cfg = test_cfg(&tc); + let backend = StartupBackend::without_match(RawStartupRead::Orders(Vec::new())) + .with_offer_latch(false) + .with_offer_outcome(Some(SellOfferOutcome::Matched)); + let note = LocalNote::generate(); + let (_dir, cursor) = temp_cursor_path("publication-immediate-match"); + + let startup = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .unwrap(); + assert_eq!( + startup, + SellerOfferStartup::Posted { + outcome: Some(SellOfferOutcome::Matched) + } + ); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 1); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_none()); + } + + #[tokio::test] + async fn publication_marker_never_posts_over_an_existing_exact_sell() { + let tc = chain_address('3'); + let owner = chain_address('4'); + let cfg = test_cfg(&tc); + let existing = raw_sell( + 99, + &owner, + Some(&tc), + u128::from(cfg.price_per_tick), + u128::from(cfg.max_ticks), + ); + let backend = StartupBackend::without_match(RawStartupRead::Orders(vec![existing])) + .with_offer_latch(true); + let note = LocalNote::generate(); + let (_dir, cursor) = temp_cursor_path("publication-duplicate-sell"); + + let startup = prepare_seller_offer_with_persisted_submission( + ¬e, + &backend, + &cfg, + Some(&owner), + &cursor, + ) + .await + .unwrap(); + assert_eq!(startup, SellerOfferStartup::ResumedResting { order_id: 99 }); + assert_eq!(backend.post_calls.load(Ordering::Relaxed), 0); + assert!(pending_offer_submission(&cursor, &tc).unwrap().is_none()); + } + + /// Issues **#839 / #723** at a money-path consumer, where the equality DECIDES something. + /// + /// The helpers that drop the dapp id — `to_chain_param`, `normalize_wallet_address`, + /// `parse_chain_address` — are all documented to yield the workchain form, so asserting that /// they preserve it would be asserting against intended behaviour. The defect is not the /// conversion; it is that consumers then treat two DIFFERENT accounts as one. - + /// /// [`validate_resting_offer`] (`seller/mod.rs:508`) is such a consumer, and the decision it /// makes is whether this seller ADOPTS a resting SELL as its own. It normalises both sides /// through `normalize_wallet_address` (`core/src/wallet.rs:20-24`, which is /// `CanonicalAddress::parse(..).legacy()`) and compares: - + /// /// ```text - /// if actual_tc != wanted_tc { return Err(..) } //:546 + /// if actual_tc != wanted_tc { return Err(..) } // :546 /// if actual_owner != wanted_owner { return Err(..) } // the same shape, for the owner note /// ``` - + /// /// Because both sides collapse to `0:`, a resting order belonging to an account in - /// a DIFFERENT DApp -- a different contract, with different code and different money -- compares + /// a DIFFERENT DApp — a different contract, with different code and different money — compares /// equal to this seller's own. `inspect_seller_offer` (`:650`) then classifies it `Resting` and /// `prepare_seller_offer` resumes it INSTEAD of posting: the seller ends up watching, and later /// serving a match against, a TokenContract that is not its own. - + /// /// Both operands are swept, because they are separate `require`-shaped comparisons and fixing /// one would leave the other: a foreign-dapp TOKEN CONTRACT, and a foreign-dapp OWNER NOTE. - - /// Asserted on the verdict `validate_resting_offer` returns -- the production decision -- not on + /// + /// Asserted on the verdict `validate_resting_offer` returns — the production decision — not on /// the rendering of any address. - + /// /// RED on this head, for both operands. #[test] - #[ignore = "issues : address identity collapses to the account id, so a foreign-dapp \ + #[ignore = "issues #839/#723: address identity collapses to the account id, so a foreign-dapp \ TokenContract or owner note is adopted as this seller's own. RED until the code PR \ carries the dapp id into the comparison. UN-IGNORE there."] fn a_foreign_dapp_order_is_not_adopted_as_this_sellers_own() { @@ -1788,7 +2785,7 @@ mod tests { }; let ours_owner = format!("0:{owner_account}"); - // Operand 1 -- the order's TokenContract lives in another DApp. + // Operand 1 — the order's TokenContract lives in another DApp. let foreign_tc = format!("{foreign_dapp}::{account}"); let order = raw_sell( 1, @@ -1800,13 +2797,13 @@ mod tests { let verdict = validate_resting_offer(&order, Some(&ours_owner), &cfg); assert!( verdict.is_err(), - "a resting SELL on TokenContract {foreign_tc} -- a different DApp from this seller's \ - {} -- was adopted as this seller's own; the seller will watch and serve a contract it \ + "a resting SELL on TokenContract {foreign_tc} — a different DApp from this seller's \ + {} — was adopted as this seller's own; the seller will watch and serve a contract it \ does not own", cfg.token_contract ); - // Operand 2 -- the order's OWNER NOTE lives in another DApp. + // Operand 2 — the order's OWNER NOTE lives in another DApp. let foreign_owner = format!("{foreign_dapp}::{owner_account}"); let order = raw_sell( 2, @@ -1818,13 +2815,13 @@ mod tests { let verdict = validate_resting_offer(&order, Some(&ours_owner), &cfg); assert!( verdict.is_err(), - "a resting SELL owned by note {foreign_owner} -- a different DApp from this seller's \ - {ours_owner} -- was accepted as this seller's own order" + "a resting SELL owned by note {foreign_owner} — a different DApp from this seller's \ + {ours_owner} — was accepted as this seller's own order" ); // The controls, and they carry the property under test. A control that is only ever // written in the LEGACY `0:` form shares nothing with the negatives above, so the - // pair would pass on a change that rejected every canonical `::` address -- including valid + // pair would pass on a change that rejected every canonical `::` address — including valid // same-DApp orders. That change is a plausible over-correction of exactly this defect, so // the accepted cases must include the canonical form of THIS seller's own DApp. let canonical_tc = format!("{}::{account}", dexdo_core::DEXDO_DAPP_ID); @@ -1878,7 +2875,7 @@ mod tests { validate_resting_offer(&order, Some(&expected_owner), &cfg).unwrap_or_else(|error| { panic!( "{label}: a same-DApp order was refused, so the rejection above cannot be \ - attributed to the FOREIGN dapp -- it rejects the canonical form as such: \ + attributed to the FOREIGN dapp — it rejects the canonical form as such: \ {error}" ) }); @@ -1930,9 +2927,9 @@ mod tests { "expected `{expected_error}` in `{error}`" ); // Identify the TC by its ACCOUNT ID, not by one spelling of the address. The error renders - // the canonical `::` form now while `cfg.token_contract` holds the + // the canonical `::` form now (#723) while `cfg.token_contract` holds the // legacy `0:` one, so a whole-string match asserted the rendering rather than the - // identification -- and broke the moment the rendering moved, which is exactly what it did. + // identification — and broke the moment the rendering moved, which is exactly what it did. // The account id is common to every form the client can print, so this survives the next // rendering change too. let tc_account = cfg.token_contract.trim_start_matches("0:"); @@ -2240,7 +3237,7 @@ mod tests { .expect_err("terminal zero-deposit TC must fail startup"); let error = error.to_string(); - // Account id, not one spelling of the address -- the message is canonical now. + // Account id, not one spelling of the address — the message is canonical now (#723). assert!(error.contains(tc.trim_start_matches("0:")), "{error}"); assert!(error.contains("deposit=0"), "{error}"); assert!(