@@ -481,6 +481,122 @@ impl std::fmt::Display for CommitError {
481481
482482impl std:: error:: Error for CommitError { }
483483
484+ /// Why [`execute_txn_with_retry`] gave up without committing.
485+ #[ derive( Debug ) ]
486+ pub enum RetryError {
487+ /// The closure's SPARQL update failed (parse error or a failing operation). This is
488+ /// never retried — retry only applies to a commit-time [`CommitError::Conflict`], not
489+ /// to a broken update, which would fail identically on every attempt.
490+ Update ( String ) ,
491+ /// Every attempt hit [`CommitError::Conflict`]; `last` is the most recent one.
492+ Exhausted { attempts : u32 , last : CommitError } ,
493+ }
494+
495+ impl std:: fmt:: Display for RetryError {
496+ fn fmt ( & self , f : & mut std:: fmt:: Formatter < ' _ > ) -> std:: fmt:: Result {
497+ match self {
498+ RetryError :: Update ( e) => write ! ( f, "transaction update failed: {e}" ) ,
499+ RetryError :: Exhausted { attempts, last } => {
500+ write ! ( f, "transaction commit did not converge after {attempts} attempt(s): {last}" )
501+ }
502+ }
503+ }
504+ }
505+
506+ impl std:: error:: Error for RetryError { }
507+
508+ /// Runs `f` against a fresh [`WriteTxn`], retrying on [`CommitError::Conflict`] up to
509+ /// `max_attempts` total tries with a caller-supplied delay between attempts.
510+ ///
511+ /// `f` applies statements via [`WriteTxn::update`] and must NOT call `commit`/`rollback`
512+ /// itself — this function owns the transaction lifecycle: on `Ok(())` from `f` it commits;
513+ /// on a conflict it discards the attempt's working copy and retries against the new
514+ /// committed state; on `Err` from `f` it rolls back and returns immediately without
515+ /// retrying (a broken update fails the same way every time).
516+ ///
517+ /// `backoff(attempt)` is called before every retry (`attempt` is the 1-indexed number of
518+ /// the attempt about to run, so it is never called before the first) and its return value
519+ /// is passed to `std::thread::sleep`. Pass [`decorrelated_jitter_backoff`] for production
520+ /// use, or `|_| std::time::Duration::ZERO` in tests.
521+ ///
522+ /// Per the job-queue coordination guidance this helper exists for (claim-style updates
523+ /// against a single-sequenced-writer store): prefer claiming **one row per transaction**
524+ /// over a batch — conflict detection is per-triple, so disjoint single-row claims from
525+ /// different callers never conflict with each other, and randomizing candidate order
526+ /// across callers further reduces repeat collisions on the same hot row.
527+ ///
528+ /// # Panics
529+ /// Panics if `max_attempts` is 0 (there must be at least one attempt).
530+ pub fn execute_txn_with_retry < F > (
531+ manager : & TransactionManager ,
532+ max_attempts : u32 ,
533+ mut backoff : impl FnMut ( u32 ) -> std:: time:: Duration ,
534+ mut f : F ,
535+ ) -> Result < u64 , RetryError >
536+ where
537+ F : FnMut ( & mut WriteTxn < ' _ > ) -> Result < ( ) , String > ,
538+ {
539+ assert ! ( max_attempts >= 1 , "execute_txn_with_retry: max_attempts must be at least 1" ) ;
540+ let mut last_conflict = None ;
541+ for attempt in 1 ..=max_attempts {
542+ if attempt > 1 {
543+ std:: thread:: sleep ( backoff ( attempt) ) ;
544+ }
545+ let mut txn = manager. begin_write ( ) ;
546+ if let Err ( e) = f ( & mut txn) {
547+ txn. rollback ( ) ;
548+ return Err ( RetryError :: Update ( e) ) ;
549+ }
550+ match txn. commit ( ) {
551+ Ok ( version) => return Ok ( version) ,
552+ Err ( conflict @ CommitError :: Conflict { .. } ) => {
553+ last_conflict = Some ( conflict) ;
554+ }
555+ }
556+ }
557+ Err ( RetryError :: Exhausted {
558+ attempts : max_attempts,
559+ last : last_conflict. expect ( "loop runs at least once when max_attempts >= 1" ) ,
560+ } )
561+ }
562+
563+ /// A dependency-free decorrelated-jitter backoff generator (AWS Architecture Blog's
564+ /// "Exponential Backoff and Jitter"), for use with [`execute_txn_with_retry`].
565+ ///
566+ /// Each call returns `min(cap, random_between(base, previous * 3))`, starting from `base`.
567+ /// Randomness comes from a splitmix64 step seeded from the wall clock and a per-call atomic
568+ /// counter — not cryptographic, but sufficient to decorrelate concurrent callers' retry
569+ /// timing, and it keeps the opt-in `txn` feature dependency-free (no `rand` crate).
570+ pub fn decorrelated_jitter_backoff (
571+ base : std:: time:: Duration ,
572+ cap : std:: time:: Duration ,
573+ ) -> impl FnMut ( u32 ) -> std:: time:: Duration {
574+ use std:: sync:: atomic:: { AtomicU64 , Ordering } ;
575+ static COUNTER : AtomicU64 = AtomicU64 :: new ( 0 ) ;
576+
577+ fn next_u64 ( ) -> u64 {
578+ let seed = std:: time:: SystemTime :: now ( )
579+ . duration_since ( std:: time:: UNIX_EPOCH )
580+ . map ( |d| d. as_nanos ( ) as u64 )
581+ . unwrap_or ( 0 )
582+ ^ COUNTER . fetch_add ( 1 , Ordering :: Relaxed ) . wrapping_mul ( 0x9E3779B97F4A7C15 ) ;
583+ // splitmix64 finalizer
584+ let mut z = seed. wrapping_add ( 0x9E3779B97F4A7C15 ) ;
585+ z = ( z ^ ( z >> 30 ) ) . wrapping_mul ( 0xBF58476D1CE4E5B9 ) ;
586+ z = ( z ^ ( z >> 27 ) ) . wrapping_mul ( 0x94D049BB133111EB ) ;
587+ z ^ ( z >> 31 )
588+ }
589+
590+ let mut previous = base;
591+ move |_attempt| {
592+ let upper = previous. saturating_mul ( 3 ) . max ( base) ;
593+ let span = upper. saturating_sub ( base) . as_nanos ( ) . max ( 1 ) as u64 ;
594+ let jittered = base + std:: time:: Duration :: from_nanos ( next_u64 ( ) % span) ;
595+ previous = jittered. min ( cap) ;
596+ previous
597+ }
598+ }
599+
484600#[ cfg( test) ]
485601mod tests {
486602 use super :: * ;
@@ -522,4 +638,83 @@ mod tests {
522638 let r = m. begin_read ( ) ;
523639 assert_eq ! ( crate :: count( & r, "SELECT * WHERE { ?s ?p ?o }" ) . unwrap( ) , 2 ) ;
524640 }
641+
642+ /// Forces a guaranteed write-write conflict against `txn`: `CLEAR DEFAULT` always
643+ /// records a slot-level `Default` write key (unlike a duplicate `INSERT DATA`, which
644+ /// may resolve to an empty, non-conflicting write set), and a default-graph triple
645+ /// insert always conflicts with it (`WriteKey::conflicts`).
646+ fn commit_a_conflicting_clear ( m : & TransactionManager ) {
647+ let mut interloper = m. begin_write ( ) ;
648+ interloper. update ( "CLEAR DEFAULT" ) . unwrap ( ) ;
649+ interloper. commit ( ) . unwrap ( ) ;
650+ }
651+
652+ #[ test]
653+ fn execute_txn_with_retry_retries_once_then_succeeds ( ) {
654+ let m = TransactionManager :: new ( g ( "@prefix : <http://ex/> . :a :p :b ." ) ) ;
655+ let mut attempts = 0 ;
656+ let result = execute_txn_with_retry ( & m, 3 , |_| std:: time:: Duration :: ZERO , |txn| {
657+ attempts += 1 ;
658+ if attempts == 1 {
659+ // A concurrent writer commits an overlapping change between this
660+ // transaction's begin and its own commit, forcing exactly one conflict.
661+ commit_a_conflicting_clear ( & m) ;
662+ }
663+ txn. update ( "PREFIX : <http://ex/> INSERT DATA { :x :p :y }" )
664+ } ) ;
665+ assert_eq ! ( attempts, 2 , "should retry exactly once after the forced conflict" ) ;
666+ assert ! ( result. is_ok( ) , "should converge on the second attempt: {result:?}" ) ;
667+ }
668+
669+ #[ test]
670+ fn execute_txn_with_retry_exhausts_after_max_attempts ( ) {
671+ let m = TransactionManager :: new ( g ( "@prefix : <http://ex/> . :a :p :b ." ) ) ;
672+ let mut attempts = 0 ;
673+ let result = execute_txn_with_retry ( & m, 3 , |_| std:: time:: Duration :: ZERO , |txn| {
674+ attempts += 1 ;
675+ // Every attempt loses the race, so this can never converge.
676+ commit_a_conflicting_clear ( & m) ;
677+ txn. update ( "PREFIX : <http://ex/> INSERT DATA { :x :p :y }" )
678+ } ) ;
679+ assert_eq ! ( attempts, 3 ) ;
680+ match result {
681+ Err ( RetryError :: Exhausted { attempts : 3 , .. } ) => { }
682+ other => panic ! ( "expected Exhausted after 3 attempts, got {other:?}" ) ,
683+ }
684+ }
685+
686+ #[ test]
687+ fn execute_txn_with_retry_does_not_retry_an_update_error ( ) {
688+ let m = TransactionManager :: new ( g ( "@prefix : <http://ex/> . :a :p :b ." ) ) ;
689+ let mut attempts = 0 ;
690+ let result = execute_txn_with_retry ( & m, 5 , |_| std:: time:: Duration :: ZERO , |_txn| {
691+ attempts += 1 ;
692+ Err ( "boom" . to_string ( ) )
693+ } ) ;
694+ assert_eq ! ( attempts, 1 , "an update error must not be retried" ) ;
695+ match result {
696+ Err ( RetryError :: Update ( msg) ) => assert_eq ! ( msg, "boom" ) ,
697+ other => panic ! ( "expected Update(\" boom\" ), got {other:?}" ) ,
698+ }
699+ }
700+
701+ #[ test]
702+ #[ should_panic( expected = "max_attempts must be at least 1" ) ]
703+ fn execute_txn_with_retry_panics_on_zero_attempts ( ) {
704+ let m = TransactionManager :: new ( g ( "@prefix : <http://ex/> . :a :p :b ." ) ) ;
705+ let _ = execute_txn_with_retry ( & m, 0 , |_| std:: time:: Duration :: ZERO , |_| Ok ( ( ) ) ) ;
706+ }
707+
708+ #[ test]
709+ fn decorrelated_jitter_backoff_stays_within_base_and_cap ( ) {
710+ use std:: time:: Duration ;
711+ let base = Duration :: from_millis ( 10 ) ;
712+ let cap = Duration :: from_millis ( 200 ) ;
713+ let mut backoff = decorrelated_jitter_backoff ( base, cap) ;
714+ for attempt in 1 ..=20u32 {
715+ let d = backoff ( attempt) ;
716+ assert ! ( d >= base, "{d:?} below base {base:?}" ) ;
717+ assert ! ( d <= cap, "{d:?} above cap {cap:?}" ) ;
718+ }
719+ }
525720}
0 commit comments