@@ -101,6 +101,17 @@ impl StorageBackend {
101101 } ,
102102 }
103103 }
104+
105+ /// What a conditional write sends on the wire, named for an operator
106+ /// reading a probe failure. The dialects use different headers, so a
107+ /// message that names only one sends half the fleet looking in the
108+ /// wrong place.
109+ fn precondition ( self ) -> & ' static str {
110+ match self {
111+ StorageBackend :: S3 => "If-Match / If-None-Match" ,
112+ StorageBackend :: Gcs => "x-goog-if-generation-match" ,
113+ }
114+ }
104115}
105116
106117/// One object-store bucket, optionally scoped to a key prefix. Cheap to
@@ -539,6 +550,135 @@ impl Bucket {
539550 }
540551 }
541552 }
553+
554+ /// Run the conditional-write contract against the live bucket.
555+ ///
556+ /// A store can accept a precondition header and then ignore it, and no
557+ /// capability API answers whether it does. So the probe provokes the
558+ /// two rejections a conforming store must produce, and checks it
559+ /// produced them. celld decides which node owns a cell with a
560+ /// conditional write, so a store that applies a write it must reject
561+ /// lets two nodes own one cell (denoland/celld#137).
562+ ///
563+ /// The two outcomes are separated because they need different
564+ /// responses. A `Violation` is a property of the store and never
565+ /// clears, so it can stop a node. An `Err` is ambiguous — a network
566+ /// fault, a rejected credential — and a retry can clear it, so a
567+ /// caller that must not fail on a transient blip keeps serving.
568+ pub ( crate ) async fn probe_cas_steps ( & self ) -> anyhow:: Result < CasVerdict > {
569+ let nanos = std:: time:: SystemTime :: now ( )
570+ . duration_since ( std:: time:: UNIX_EPOCH )
571+ . map ( |since| since. as_nanos ( ) )
572+ . unwrap_or_default ( ) ;
573+ // Unique per probe, so several nodes probing at once touch
574+ // disjoint keys and no probe reads another one's object as the
575+ // store misbehaving — a collision surfaces as a false `Violation`,
576+ // which stops a node. The random half carries that alone, because
577+ // a container fleet shares pid 1 and a clock before the epoch
578+ // leaves `nanos` at zero.
579+ let key = format ! (
580+ "probe/cas-{nanos}-{}-{:016x}" ,
581+ std:: process:: id( ) ,
582+ rand:: random:: <u64 >( )
583+ ) ;
584+ let verdict = self . cas_contract ( & key) . await ;
585+ // The object is debris on every path, so retire it before the
586+ // verdict. A delete that fails leaves one tiny object under
587+ // `probe/`, which nothing lists and nothing reads — but a
588+ // credential that cannot delete accrues one per boot, so say so.
589+ if let Err ( error) = self . delete ( & key) . await {
590+ tracing:: warn!( %error, "the conditional-write probe could not delete its object" ) ;
591+ }
592+ verdict
593+ }
594+
595+ /// [`Self::probe_cas_steps`] collapsed to one answer, where any wrong
596+ /// answer fails the check.
597+ pub async fn probe_cas ( & self ) -> anyhow:: Result < ( ) > {
598+ match self . probe_cas_steps ( ) . await ? {
599+ CasVerdict :: Conformant => Ok ( ( ) ) ,
600+ CasVerdict :: Violation ( reason) => Err ( anyhow ! ( reason) ) ,
601+ }
602+ }
603+
604+ /// The four steps, against one key. Steps 2 and 4 must be rejected;
605+ /// a store that applies either one cannot fence.
606+ async fn cas_contract ( & self , key : & str ) -> anyhow:: Result < CasVerdict > {
607+ let precondition = self . backend . precondition ( ) ;
608+ let ambiguous = || {
609+ format ! (
610+ "the store answered a conditional write with an error where celld requires a \
611+ \
612+
613+ )
614+ } ;
615+
616+ // 1. A create on an absent key applies, and answers the token that
617+ // steps 3 and 4 need.
618+ let Some ( token) = self
619+ . put_cas ( key, b"probe-create" . to_vec ( ) , None )
620+ . await
621+ . context ( "the conditional-write probe could not create its object" ) ?
622+ else {
623+ return Ok ( CasVerdict :: Violation (
624+ "the store rejected a conditional create of an object that does not exist"
625+ . to_string ( ) ,
626+ ) ) ;
627+ } ;
628+
629+ // 2. A create over the object step 1 wrote must be rejected.
630+ if self
631+ . put_cas ( key, b"probe-recreate" . to_vec ( ) , None )
632+ . await
633+ . with_context ( ambiguous) ?
634+ . is_some ( )
635+ {
636+ return Ok ( CasVerdict :: Violation ( format ! (
637+ "the store overwrote an object although the write was conditional on that object \
638+ \
639+
640+ ) ) ) ;
641+ }
642+
643+ // 3. An update that carries the current token applies, and that
644+ // retires the token step 4 reuses.
645+ if self
646+ . put_cas ( key, b"probe-update" . to_vec ( ) , Some ( & token) )
647+ . await
648+ . context ( "the conditional-write probe could not update its object" ) ?
649+ . is_none ( )
650+ {
651+ return Ok ( CasVerdict :: Violation (
652+ "the store rejected a conditional update that carried the current token"
653+ . to_string ( ) ,
654+ ) ) ;
655+ }
656+
657+ // 4. The token is stale now, so the update must be rejected. This
658+ // step is the fencing contract itself.
659+ if self
660+ . put_cas ( key, b"probe-stale" . to_vec ( ) , Some ( & token) )
661+ . await
662+ . with_context ( ambiguous) ?
663+ . is_some ( )
664+ {
665+ return Ok ( CasVerdict :: Violation ( format ! (
666+ "the store applied a conditional write that carried a stale token; the store \
667+
668+ ) ) ) ;
669+ }
670+
671+ Ok ( CasVerdict :: Conformant )
672+ }
673+ }
674+
675+ /// What [`Bucket::probe_cas_steps`] found.
676+ pub ( crate ) enum CasVerdict {
677+ /// The store rejected both writes it had to reject.
678+ Conformant ,
679+ /// The store answered wrongly, and the string says how. This never
680+ /// clears on a retry, so a caller can act on it.
681+ Violation ( String ) ,
542682}
543683
544684/// The replica-lane store for a `gs://` fleet bucket: its own transport
@@ -613,43 +753,15 @@ mod live_cas {
613753 } ) ;
614754 let bucket = Bucket :: open ( & name, endpoint. as_deref ( ) , & region, creds, Some ( "cas-test" ) )
615755 . expect ( "open bucket" ) ;
616- let nanos = std:: time:: SystemTime :: now ( )
617- . duration_since ( std:: time:: UNIX_EPOCH )
618- . unwrap ( )
619- . as_nanos ( ) ;
620- let key = format ! ( "cas-probe/{nanos}" ) ;
621756
622- // 1. Create on an absent key applies.
623- let e1 = bucket
624- . put_cas ( & key, b"v1" . to_vec ( ) , None )
625- . await
626- . expect ( "create must not error" )
627- . expect ( "fresh create must apply (Ok(Some))" ) ;
628- // 2. Create over an existing key is cleanly rejected.
629- assert ! (
630- bucket
631- . put_cas( & key, b"v1b" . to_vec( ) , None )
632- . await
633- . expect( "create-again must not error" )
634- . is_none( ) ,
635- "create over an existing key must be Ok(None)"
636- ) ;
637- // 3. Update with the current etag applies.
757+ // The four steps live in `Bucket::probe_cas`, which `celld
758+ // diagnose` and node startup run against an operator's bucket.
759+ // This test points the same code at a real provider, which is the
760+ // one question a mock cannot answer.
638761 bucket
639- . put_cas ( & key , b"v3" . to_vec ( ) , Some ( & e1 ) )
762+ . probe_cas ( )
640763 . await
641- . expect ( "update must not error" )
642- . expect ( "update with current etag must apply (Ok(Some))" ) ;
643- // 4. Update with the now-stale etag is cleanly rejected — the fencing case.
644- assert ! (
645- bucket
646- . put_cas( & key, b"v4" . to_vec( ) , Some ( & e1) )
647- . await
648- . expect( "stale update must not error" )
649- . is_none( ) ,
650- "update with a stale etag must be Ok(None) — the fencing contract"
651- ) ;
652- bucket. delete ( & key) . await . expect ( "cleanup delete" ) ;
764+ . expect ( "the store must keep the conditional-write contract" ) ;
653765 eprintln ! ( "CAS verified on {name}: create / reject-create / update / reject-stale" ) ;
654766 }
655767}
0 commit comments