@@ -175,15 +175,21 @@ pub struct Seal {
175175 signing : SigningPayload ,
176176 timestamp : Option < TimestampAuthority > ,
177177 allowed_clients : Vec < String > ,
178- /// Held while one caller probes a signer that is not ready, so a burst
179- /// of requests sends the provider one probe.
180- recovery_probe : tokio:: sync:: Mutex < ( ) > ,
178+ /// Held by the recovery probe in flight, so a burst of requests sends
179+ /// the provider one probe at a time. Its value is when the last probe
180+ /// that failed ended; see [`Seal::start_signer_probe`].
181+ recovery_probe : Arc < tokio:: sync:: Mutex < Option < tokio:: time:: Instant > > > ,
181182}
182183
183184/// What a recovery probe signs: never a document, so a probe signature
184185/// cannot stand in for a seal.
185186const SIGNER_PROBE_MESSAGE : & [ u8 ] = b"registry-render-seal-signer-probe-v1" ;
186187
188+ /// How long after a failed recovery probe ends the next may start, so a
189+ /// signer that stays not ready is probed at most once a second, however
190+ /// often it is asked.
191+ const PROBE_SPACING : std:: time:: Duration = std:: time:: Duration :: from_secs ( 1 ) ;
192+
187193impl std:: fmt:: Debug for Seal {
188194 fn fmt ( & self , f : & mut std:: fmt:: Formatter < ' _ > ) -> std:: fmt:: Result {
189195 f. debug_struct ( "Seal" )
@@ -443,7 +449,7 @@ impl Seal {
443449 contents_capacity,
444450 timestamp : None ,
445451 allowed_clients : vec ! [ "records-service" . to_owned( ) ] ,
446- recovery_probe : tokio:: sync:: Mutex :: new ( ( ) ) ,
452+ recovery_probe : Arc :: new ( tokio:: sync:: Mutex :: new ( None ) ) ,
447453 }
448454 }
449455}
@@ -755,7 +761,7 @@ impl Seal {
755761 contents_capacity,
756762 timestamp,
757763 allowed_clients : seal. allowed_clients . as_slice ( ) . to_vec ( ) ,
758- recovery_probe : tokio:: sync:: Mutex :: new ( ( ) ) ,
764+ recovery_probe : Arc :: new ( tokio:: sync:: Mutex :: new ( None ) ) ,
759765 } )
760766 }
761767
@@ -799,26 +805,64 @@ impl Seal {
799805 . any ( |client| client == client_id)
800806 }
801807
802- /// Give a signer that said it is not ready one chance to recover: one
803- /// probe signature, which the signer verifies and records in its
804- /// readiness. A Transit signer turns not ready when a call fails, and
805- /// only a later call that succeeds turns it back, so the seal route and
806- /// `/ready` probe before they ask [`Seal::availability`]. A ready signer
807- /// is never called, and concurrent callers share one probe.
808+ /// Give a signer that said it is not ready a chance to recover by
809+ /// starting a probe: one probe signature, which the signer verifies and
810+ /// records in its readiness. A Transit signer turns not ready when a
811+ /// call fails, and only a later call that succeeds turns it back, so the
812+ /// seal route and `/ready` start a probe before they ask
813+ /// [`Seal::availability`].
814+ ///
815+ /// The probe runs as its own task and holds the probe lock until it
816+ /// ends, so a caller that goes away cannot release it. A probe starts
817+ /// only when the signer reports not ready, no probe is in flight, and
818+ /// [`PROBE_SPACING`] has passed since the last probe that failed ended.
819+ /// A probe that recovers the signer clears that end, so the first probe
820+ /// after the signer turns not ready again starts at once. Returns the
821+ /// probe's handle; dropping it leaves the probe running.
822+ pub fn start_signer_probe ( & self ) -> Option < tokio:: task:: JoinHandle < ( ) > > {
823+ let Ok ( mut failed_at) = Arc :: clone ( & self . recovery_probe ) . try_lock_owned ( ) else {
824+ return None ;
825+ } ;
826+ // Read under the lock: a probe that just finished has set both.
827+ if self . signer . readiness ( ) != KeyReadiness :: NotReady
828+ || failed_at. is_some_and ( |ended| ended. elapsed ( ) < PROBE_SPACING )
829+ {
830+ return None ;
831+ }
832+ let signer = Arc :: clone ( & self . signer ) ;
833+ let seal_id = self . id . clone ( ) ;
834+ Some ( tokio:: spawn ( async move {
835+ // The signer has verified the probe signature and recorded its
836+ // readiness; the signature itself is not needed.
837+ * failed_at = match signer. sign ( SIGNER_PROBE_MESSAGE ) . await {
838+ Ok ( _) => None ,
839+ Err ( error) => {
840+ tracing:: warn!(
841+ target: "registry_render::seal" ,
842+ seal_id = %seal_id,
843+ %error,
844+ "seal signer is still not ready"
845+ ) ;
846+ Some ( tokio:: time:: Instant :: now ( ) )
847+ }
848+ } ;
849+ } ) )
850+ }
851+
852+ /// Start a probe as [`Seal::start_signer_probe`] does and wait for it to
853+ /// end, so the seal route reads [`Seal::availability`] after it. A ready
854+ /// signer is never called, and a caller that finds a probe in flight or
855+ /// held off does not wait.
808856 pub async fn recover_signer ( & self ) {
809- let Ok ( _probe ) = self . recovery_probe . try_lock ( ) else {
857+ let Some ( probe ) = self . start_signer_probe ( ) else {
810858 return ;
811859 } ;
812- // Read under the lock: a probe that just finished has set it.
813- if self . signer . readiness ( ) != KeyReadiness :: NotReady {
814- return ;
815- }
816- if let Err ( error) = self . signer . sign ( SIGNER_PROBE_MESSAGE ) . await {
817- tracing:: warn!(
860+ if let Err ( error) = probe. await {
861+ tracing:: error!(
818862 target: "registry_render::seal" ,
819863 seal_id = %self . id,
820- % error,
821- "seal signer is still not ready "
864+ panicked = error. is_panic ( ) ,
865+ "seal signer probe did not finish "
822866 ) ;
823867 }
824868 }
@@ -1559,12 +1603,14 @@ mod tests {
15591603 }
15601604
15611605 /// A signer whose readiness follows its last signature, as a Transit
1562- /// signer's does, and whose signatures succeed while `healthy` is set.
1606+ /// signer's does, whose signatures succeed while `healthy` is set, and
1607+ /// each of whose calls takes `takes`.
15631608 struct ProbedSigner {
15641609 inner : LocalJwkSigner ,
15651610 healthy : std:: sync:: atomic:: AtomicBool ,
15661611 ready : std:: sync:: atomic:: AtomicBool ,
15671612 calls : AtomicUsize ,
1613+ takes : std:: time:: Duration ,
15681614 }
15691615
15701616 #[ async_trait:: async_trait]
@@ -1591,7 +1637,7 @@ mod tests {
15911637
15921638 async fn sign ( & self , payload : & [ u8 ] ) -> Result < Vec < u8 > , SigningError > {
15931639 self . calls . fetch_add ( 1 , Ordering :: SeqCst ) ;
1594- tokio:: time:: sleep ( std :: time :: Duration :: from_millis ( 20 ) ) . await ;
1640+ tokio:: time:: sleep ( self . takes ) . await ;
15951641 let healthy = self . healthy . load ( Ordering :: SeqCst ) ;
15961642 self . ready . store ( healthy, Ordering :: SeqCst ) ;
15971643 if healthy {
@@ -1602,27 +1648,36 @@ mod tests {
16021648 }
16031649 }
16041650
1605- /// A signer that turned not ready gets one probe per recovery, shared by
1606- /// concurrent callers, and the seal is available again once the probe
1607- /// signs; a ready signer is never probed.
1608- #[ tokio:: test]
1609- async fn a_signer_that_turned_not_ready_is_probed_once_and_recovers ( ) {
1651+ /// A seal whose signer is not ready and fails until it is made
1652+ /// healthy, each call taking `takes`.
1653+ fn probed_seal ( takes : std:: time:: Duration ) -> ( Arc < ProbedSigner > , Arc < Seal > ) {
16101654 let pki = TestPki :: generate ( ) . expect ( "test PKI" ) ;
16111655 let test_seal = pki. issue_seal ( SigningAlgorithm :: Es256 ) . expect ( "test seal" ) ;
16121656 let signer = Arc :: new ( ProbedSigner {
16131657 inner : test_seal. signer ( ) . expect ( "test signer" ) ,
16141658 healthy : std:: sync:: atomic:: AtomicBool :: new ( false ) ,
16151659 ready : std:: sync:: atomic:: AtomicBool :: new ( false ) ,
16161660 calls : AtomicUsize :: new ( 0 ) ,
1661+ takes,
16171662 } ) ;
16181663 let seal = Seal :: for_test ( "ministry" , & test_seal, Arc :: clone ( & signer) as _ ) ;
1664+ ( signer, Arc :: new ( seal) )
1665+ }
1666+
1667+ /// A signer that turned not ready gets one probe at a time, shared by
1668+ /// concurrent callers, and the seal is available again once a probe
1669+ /// signs; a ready signer is never probed.
1670+ #[ tokio:: test( start_paused = true ) ]
1671+ async fn a_signer_that_turned_not_ready_is_probed_once_and_recovers ( ) {
1672+ let ( signer, seal) = probed_seal ( std:: time:: Duration :: from_millis ( 20 ) ) ;
16191673 let now = OffsetDateTime :: now_utc ( ) ;
16201674
16211675 tokio:: join!( seal. recover_signer( ) , seal. recover_signer( ) ) ;
16221676 assert_eq ! ( signer. calls. load( Ordering :: SeqCst ) , 1 , "one shared probe" ) ;
16231677 assert_eq ! ( seal. availability( now) , Err ( Unavailable :: SignerNotReady ) ) ;
16241678
16251679 signer. healthy . store ( true , Ordering :: SeqCst ) ;
1680+ tokio:: time:: advance ( PROBE_SPACING ) . await ;
16261681 seal. recover_signer ( ) . await ;
16271682 assert_eq ! ( signer. calls. load( Ordering :: SeqCst ) , 2 ) ;
16281683 assert_eq ! ( seal. availability( now) , Ok ( ( ) ) ) ;
@@ -1635,6 +1690,85 @@ mod tests {
16351690 ) ;
16361691 }
16371692
1693+ /// A failed probe holds off the next until [`PROBE_SPACING`] after it
1694+ /// ended, not after it started, so a probe that waited out a long
1695+ /// timeout is not retried at once. A probe that recovers the signer
1696+ /// leaves nothing to wait for: the first probe after the signer turns
1697+ /// not ready again starts at once.
1698+ #[ tokio:: test( start_paused = true ) ]
1699+ async fn a_failed_probe_holds_off_the_next_from_when_it_ended ( ) {
1700+ let ( signer, seal) = probed_seal ( std:: time:: Duration :: from_secs ( 30 ) ) ;
1701+ let now = OffsetDateTime :: now_utc ( ) ;
1702+
1703+ seal. recover_signer ( ) . await ;
1704+ assert_eq ! ( signer. calls. load( Ordering :: SeqCst ) , 1 ) ;
1705+ assert ! (
1706+ seal. start_signer_probe( ) . is_none( ) ,
1707+ "no probe straight after a failed one"
1708+ ) ;
1709+ tokio:: time:: advance ( PROBE_SPACING - std:: time:: Duration :: from_millis ( 1 ) ) . await ;
1710+ assert ! (
1711+ seal. start_signer_probe( ) . is_none( ) ,
1712+ "no probe within the spacing of the failed probe's end"
1713+ ) ;
1714+ tokio:: time:: advance ( std:: time:: Duration :: from_millis ( 1 ) ) . await ;
1715+ signer. healthy . store ( true , Ordering :: SeqCst ) ;
1716+ seal. start_signer_probe ( )
1717+ . expect ( "a probe once the spacing has passed" )
1718+ . await
1719+ . expect ( "the probe ends" ) ;
1720+ assert_eq ! ( signer. calls. load( Ordering :: SeqCst ) , 2 ) ;
1721+ assert_eq ! ( seal. availability( now) , Ok ( ( ) ) ) ;
1722+
1723+ // A seal's own signature fails, and the signer turns not ready.
1724+ signer. healthy . store ( false , Ordering :: SeqCst ) ;
1725+ assert ! ( signer. sign( b"a document digest" ) . await . is_err( ) ) ;
1726+ assert_eq ! ( seal. availability( now) , Err ( Unavailable :: SignerNotReady ) ) ;
1727+ signer. healthy . store ( true , Ordering :: SeqCst ) ;
1728+ seal. start_signer_probe ( )
1729+ . expect ( "the first probe after the signer turns not ready is not held off" )
1730+ . await
1731+ . expect ( "the probe ends" ) ;
1732+ assert_eq ! ( signer. calls. load( Ordering :: SeqCst ) , 4 ) ;
1733+ assert_eq ! ( seal. availability( now) , Ok ( ( ) ) ) ;
1734+ }
1735+
1736+ /// The probe holds the probe lock itself. A caller that goes away
1737+ /// mid-probe, as a disconnected client's request does, neither releases
1738+ /// the lock nor stops the probe, which still records its result.
1739+ #[ tokio:: test( start_paused = true ) ]
1740+ async fn a_caller_that_goes_away_leaves_its_probe_running_and_holding_the_lock ( ) {
1741+ let ( signer, seal) = probed_seal ( std:: time:: Duration :: from_millis ( 20 ) ) ;
1742+ let now = OffsetDateTime :: now_utc ( ) ;
1743+ signer. healthy . store ( true , Ordering :: SeqCst ) ;
1744+
1745+ let caller = tokio:: spawn ( {
1746+ let seal = Arc :: clone ( & seal) ;
1747+ async move { seal. recover_signer ( ) . await }
1748+ } ) ;
1749+ tokio:: time:: sleep ( std:: time:: Duration :: from_millis ( 5 ) ) . await ;
1750+ assert_eq ! (
1751+ signer. calls. load( Ordering :: SeqCst ) ,
1752+ 1 ,
1753+ "the probe is signing"
1754+ ) ;
1755+ caller. abort ( ) ;
1756+ assert ! ( caller. await . expect_err( "the caller is gone" ) . is_cancelled( ) ) ;
1757+ assert ! (
1758+ seal. start_signer_probe( ) . is_none( ) ,
1759+ "the probe still holds the lock"
1760+ ) ;
1761+ assert_eq ! ( seal. availability( now) , Err ( Unavailable :: SignerNotReady ) ) ;
1762+
1763+ tokio:: time:: sleep ( std:: time:: Duration :: from_millis ( 20 ) ) . await ;
1764+ assert_eq ! ( signer. calls. load( Ordering :: SeqCst ) , 1 ) ;
1765+ assert_eq ! (
1766+ seal. availability( now) ,
1767+ Ok ( ( ) ) ,
1768+ "the probe finished and recorded its result"
1769+ ) ;
1770+ }
1771+
16381772 #[ test]
16391773 fn levels_and_reasons_have_stable_names ( ) {
16401774 assert_eq ! ( SealLevel :: BT . as_str( ) , "B-T" ) ;
0 commit comments