Preserve safe rejection identity through owned transaction cleanup

modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/OwnedLeaseConnectionFactory.java

@@ -71,16 +71,17 @@ public <T> Mono<RootCompletion<T>> executeBounded(BoundedR2dbcResources resource
                                 @Override public Mono<Void> afterCompletion(int state){return Mono.fromRunnable(()->{if(state==STATUS_ROLLED_BACK)owned.rollbackAcknowledged.set(true);});}
                             });
                             owned.stage.set(RootTransactionException.Stage.WORK);
-                            return Mono.defer(()->work.apply(owned)).switchIfEmpty(Mono.error(new IllegalStateException("One bounded root result required")));
+                            return Mono.defer(()->work.apply(owned)).switchIfEmpty(Mono.error(new IllegalStateException("One bounded root result required")))
+                                .doOnError(owned::observeWorkRejection);
                         }).flux()).collectList().doOnSuccess(values->owned.boundaryCompleted.set(true)).flatMap(values->
                             values.size()==1?Mono.just(values.getFirst()):Mono.error(new IllegalStateException("One bounded root result required")));
-                    return transaction.timeout(owned.deadline()).doOnError(error->owned.captureFailure());
+                    return transaction.timeout(owned.deadline()).doOnError(owned::captureFailure);
                 },Invocation::cleanup,(owned,error)->owned.cleanup(),Invocation::cleanup);
                 return bounded.flatMap(value->{
                     scope.stage.set(RootTransactionException.Stage.FINAL_BUDGET);
                     scope.remaining();
                     return Mono.just(new RootCompletion<>(value,scope.facts()));
-                }).onErrorMap(error->scope.failure()).contextWrite(c->c.put(INVOCATION,scope));
+                }).onErrorMap(scope::failure).contextWrite(c->c.put(INVOCATION,scope));
             });
         });
     }
@@ -99,16 +100,47 @@ private static final class Invocation implements RootTransactionScope {
         final AtomicReference<Mono<Void>> rollback=new AtomicReference<>();
         final AtomicBoolean incompleteCleanup=new AtomicBoolean();
         final AtomicReference<CleanupFact> cleanupFact=new AtomicReference<>(CleanupFact.UNKNOWN);
+        final Object rejectionOwner=new Object();
+        final AtomicReference<WorkRejectionSignal> registeredRejection=new AtomicReference<>(),workRejection=new AtomicReference<>(),terminalRejection=new AtomicReference<>();
         final Sinks.Many<Boolean> narrowed=Sinks.many().replay().latest();
         reactor.util.context.Context transactionContext;
         Invocation(OwnedLeaseConnectionFactory factory,BoundedR2dbcResources resources,R2dbcOperationBudget original,CleanupPolicy policy){this.factory=factory;this.resources=resources;this.original=original;this.cap=new AtomicReference<>(original);this.policy=policy;}
         @Override public Duration remaining(){var a=original.remaining();var b=cap.get().remaining();return a.compareTo(b)<0?a:b;}
         @Override public void narrow(R2dbcOperationBudget observedCap){cap.updateAndGet(current->current.intersect(observedCap));remaining();if(narrowed.tryEmitNext(true).isFailure())throw new IllegalStateException("Bounded deadline observation unavailable");}
+        @Override public <T> Mono<T> reject(RootWorkRejectionKey key){
+            return factory.requireBoundScope(resources,this).then(Mono.defer(()->{
+                remaining();var status=transaction.get();
+                if(key==null||stage.get()!=RootTransactionException.Stage.WORK||status==null||status.isCompleted()
+                        ||registeredRejection.get()!=null||!key.bind(rejectionOwner))return Mono.error(rejectionUnavailable());
+                var signal=new WorkRejectionSignal(rejectionOwner,key);
+                if(!registeredRejection.compareAndSet(null,signal))return Mono.error(rejectionUnavailable());
+                status.setRollbackOnly();remaining();return Mono.error(signal);
+            }));
+        }
         Mono<Long> deadline(){return narrowed.asFlux().startWith(false).switchMap(signal->Mono.defer(()->Mono.delay(remaining()))).next();}
         Connection own(Connection raw){var lease=new Lease(raw,this);leases.add(lease);return lease;}
         void captureFailure(){failureStage.compareAndSet(null,stage.get());}
+        void observeWorkRejection(Throwable error){
+            var expected=registeredRejection.get();
+            if(expected!=null&&error==expected)workRejection.compareAndSet(null,expected);
+        }
+        void captureFailure(Throwable error){
+            captureFailure();var expected=workRejection.get();
+            terminalRejection.set(expected!=null&&error==expected?expected:null);
+        }
         TransactionFacts facts(){return new TransactionFacts(nativeAttempted.get(),commitAcknowledged.get(),boundaryCompleted.get(),rollbackAcknowledged.get());}
         RootTransactionException failure(){return new RootTransactionException(facts(),cleanupFact.get(),Optional.ofNullable(failureStage.get()).orElse(stage.get()));}
+        RootTransactionException failure(Throwable error){
+            var observed=facts();var cleaned=cleanupFact.get();var expected=terminalRejection.get();RootWorkRejectionKey key=null;
+            var status=transaction.get();
+            if(expected!=null&&error==expected&&expected==workRejection.get()&&expected==registeredRejection.get()
+                    &&expected.owner==rejectionOwner&&expected.key.belongsTo(rejectionOwner)&&resources.current()
+                    &&status!=null&&status.isCompleted()&&observed.nativeAttempted()&&observed.rollbackAcknowledged()
+                    &&!observed.commitAcknowledged()&&!observed.boundaryCompleted()&&cleaned==CleanupFact.COMPLETE){
+                try{remaining();key=expected.key;}catch(RuntimeException expired){key=null;}
+            }
+            return new RootTransactionException(observed,cleaned,Optional.ofNullable(failureStage.get()).orElse(stage.get()),key);
+        }
         ReactiveTransactionManager manager(){return new ReactiveTransactionManager(){
             @Override public Mono<ReactiveTransaction> getReactiveTransaction(TransactionDefinition d){return Mono.deferContextual(context->{
                 if(!resources.current())return Mono.error(gateFailure());transactionContext=reactor.util.context.Context.of(context);var limit=remaining();return resources.manager().getReactiveTransaction(d).timeout(limit);
@@ -133,6 +165,12 @@ Mono<Void> cleanup(){return Mono.defer(()->{
               .then(Mono.defer(()->incomplete.get()?Mono.error(failure()):Mono.empty()));
         });}
     }
+    private static IllegalStateException rejectionUnavailable(){return new IllegalStateException("Safe owned work rejection unavailable");}
+    private static final class WorkRejectionSignal extends RuntimeException {
+        private static final long serialVersionUID=1L;
+        final transient Object owner;final transient RootWorkRejectionKey key;
+        WorkRejectionSignal(Object owner,RootWorkRejectionKey key){super("BOUNDED_ROOT_WORK_REJECTED",null,false,false);this.owner=owner;this.key=key;}
+    }
     private static final class Lease implements Connection {
         final Connection raw;final Invocation scope;final Mono<Void> close;
         Lease(Connection raw,Invocation scope){this.raw=Objects.requireNonNull(raw);this.scope=scope;close=Mono.defer(()->Mono.from(raw.close())).timeout(scope.policy.close()).cache();}

modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionException.java

@@ -7,13 +7,19 @@
 import java.util.Objects;
 public final class RootTransactionException extends IllegalStateException {
     public enum Stage { GATE, ACQUIRE, BEGIN, WORK, COMMIT, ROLLBACK, CLOSE, FINAL_BUDGET }
+    private static final long serialVersionUID=1L;
     private final TransactionFacts facts;
     private final CleanupFact cleanup;
     private final Stage stage;
+    private final transient RootWorkRejectionKey rejection;
     public RootTransactionException(TransactionFacts facts,CleanupFact cleanup,Stage stage){
-        super("BOUNDED_ROOT_TRANSACTION_UNAVAILABLE");this.facts=Objects.requireNonNull(facts);this.cleanup=Objects.requireNonNull(cleanup);this.stage=Objects.requireNonNull(stage);
+        this(facts,cleanup,stage,null);
+    }
+    RootTransactionException(TransactionFacts facts,CleanupFact cleanup,Stage stage,RootWorkRejectionKey rejection){
+        super("BOUNDED_ROOT_TRANSACTION_UNAVAILABLE");this.facts=Objects.requireNonNull(facts);this.cleanup=Objects.requireNonNull(cleanup);this.stage=Objects.requireNonNull(stage);this.rejection=rejection;
     }
     public TransactionFacts transactionFacts(){return facts;}
     public CleanupFact cleanupFact(){return cleanup;}
     public Stage stage(){return stage;}
+    public boolean workRejectedBy(RootWorkRejectionKey key){return key!=null&&rejection==key;}
 }

modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionScope.java

@@ -5,7 +5,12 @@
  */
 package org.fireflyframework.core.transaction;
 import java.time.Duration;
+import reactor.core.publisher.Mono;
 public interface RootTransactionScope {
     Duration remaining();
     void narrow(R2dbcOperationBudget observedCap);
+    /** Explicit opt-in; custom scopes cannot manufacture a successful rejection correlation. */
+    default <T> Mono<T> reject(RootWorkRejectionKey key){
+        return Mono.error(new IllegalStateException("Safe owned work rejection unavailable"));
+    }
 }

modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootWorkRejectionKey.java

@@ -0,0 +1,14 @@
+/* Copyright 2026 Firefly Software Foundation. Licensed under the Apache License, Version 2.0. */
+package org.fireflyframework.core.transaction;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Fresh per subscriber; correlates an owned rejection without carrying business data or authority. */
+public final class RootWorkRejectionKey {
+    private final AtomicReference<Object> owner=new AtomicReference<>();
+    private RootWorkRejectionKey() { }
+    public static RootWorkRejectionKey create(){return new RootWorkRejectionKey();}
+    boolean bind(Object invocation){return owner.compareAndSet(null,invocation);}
+    boolean belongsTo(Object invocation){return owner.get()==invocation;}
+    @Override public String toString(){return "RootWorkRejectionKey";}
+}