diff --git a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/OwnedLeaseConnectionFactory.java b/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/OwnedLeaseConnectionFactory.java index fdea1850..54dafb14 100644 --- a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/OwnedLeaseConnectionFactory.java +++ b/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/OwnedLeaseConnectionFactory.java @@ -71,16 +71,17 @@ public Mono> executeBounded(BoundedR2dbcResources resource @Override public Mono 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> rollback=new AtomicReference<>(); final AtomicBoolean incompleteCleanup=new AtomicBoolean(); final AtomicReference cleanupFact=new AtomicReference<>(CleanupFact.UNKNOWN); + final Object rejectionOwner=new Object(); + final AtomicReference registeredRejection=new AtomicReference<>(),workRejection=new AtomicReference<>(),terminalRejection=new AtomicReference<>(); final Sinks.Many 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 Mono 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 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 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 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 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();} diff --git a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionException.java b/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionException.java index 1b0d5f4a..f13fdbeb 100644 --- a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionException.java +++ b/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;} } diff --git a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionScope.java b/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionScope.java index f52306a6..1e3418ce 100644 --- a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootTransactionScope.java +++ b/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 Mono reject(RootWorkRejectionKey key){ + return Mono.error(new IllegalStateException("Safe owned work rejection unavailable")); + } } diff --git a/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootWorkRejectionKey.java b/modules/r2dbc/src/main/java/org/fireflyframework/core/transaction/RootWorkRejectionKey.java new file mode 100644 index 00000000..c275297a --- /dev/null +++ b/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 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";} +} diff --git a/modules/r2dbc/src/test/java/org/fireflyframework/core/transaction/RootWorkRejectionTest.java b/modules/r2dbc/src/test/java/org/fireflyframework/core/transaction/RootWorkRejectionTest.java new file mode 100644 index 00000000..5689c2e6 --- /dev/null +++ b/modules/r2dbc/src/test/java/org/fireflyframework/core/transaction/RootWorkRejectionTest.java @@ -0,0 +1,151 @@ +package org.fireflyframework.core.transaction; + +import io.r2dbc.spi.*; +import java.lang.reflect.*; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.atomic.*; +import java.util.function.*; +import org.fireflyframework.core.transaction.impl.DefaultBoundedRootTransaction; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Publisher; +import org.springframework.data.r2dbc.core.R2dbcEntityTemplate; +import org.springframework.data.r2dbc.dialect.PostgresDialect; +import org.springframework.r2dbc.connection.R2dbcTransactionManager; +import org.springframework.r2dbc.core.DatabaseClient; +import org.springframework.transaction.reactive.*; +import org.springframework.transaction.support.DefaultTransactionDefinition; +import reactor.core.publisher.*; +import reactor.test.StepVerifier; +import reactor.util.context.ContextView; +import static org.junit.jupiter.api.Assertions.*; + +class RootWorkRejectionTest { + static final class ExpectedRejection extends RuntimeException { + private static final long serialVersionUID=1L; + ExpectedRejection(){super("PRIVATE-REJECTION-CONTENT");} + } + static final class Driver implements ConnectionFactory { + final BoundedTransactionBoundaryTest.Driver original=new BoundedTransactionBoundaryTest.Driver(); + final Sinks.Empty rollbackGate=Sinks.empty(),closeGate=Sinks.empty(); + boolean holdRollback,holdClose; + final Connection connection=(Connection)Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{Connection.class},(proxy,method,args)->{ + if(method.getName().equals("rollbackTransaction"))return Mono.from(original.connection.rollbackTransaction()).then(holdRollback?rollbackGate.asMono():Mono.empty()); + if(method.getName().equals("close"))return Mono.from(original.connection.close()).then(holdClose?closeGate.asMono():Mono.empty()); + try{return method.invoke(original.connection,args);}catch(InvocationTargetException error){throw error.getCause();} + }); + public Publisher create(){return Mono.from(original.create()).map(ignored->connection);} + public ConnectionFactoryMetadata getMetadata(){return original.getMetadata();} + } + static final class Fixture { + final BoundedTransactionBoundaryTest.Time time=new BoundedTransactionBoundaryTest.Time();final Driver driver=new Driver(); + final OwnedLeaseConnectionFactory factory=new OwnedLeaseConnectionFactory(driver); + final DatabaseClient client=DatabaseClient.create(factory);final R2dbcEntityTemplate template=new R2dbcEntityTemplate(client,PostgresDialect.INSTANCE); + final R2dbcTransactionManager manager=new R2dbcTransactionManager(factory); + final BoundedRootTransaction root=new DefaultBoundedRootTransaction(BoundedR2dbcResources.verify(factory,client,template,manager),new CleanupPolicy(Duration.ofSeconds(2),Duration.ofSeconds(2))); + final R2dbcOperationBudget budget=time.budget(5000); + Mono> execute(Function> work){return root.execute(budget,new DefaultTransactionDefinition(),work);} + } + record Failed(Fixture fixture,RootTransactionException error) { } + static Failed failure(Consumer configure,BiFunction> work,long seconds){ + var fixture=new AtomicReference();var result=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();fixture.set(f);configure.accept(f);return f.execute(scope->work.apply(f,scope));}) + .thenAwait(Duration.ofSeconds(seconds)).expectErrorSatisfies(error->{var root=assertInstanceOf(RootTransactionException.class,error);result.set(root);assertEquals("BOUNDED_ROOT_TRANSACTION_UNAVAILABLE",root.getMessage());assertNull(root.getCause());assertFalse(root.toString().contains("PRIVATE"));}).verify(Duration.ofSeconds(1)); + return new Failed(fixture.get(),result.get()); + } + @Test void exactTypedRejectionIsCertifiedOnlyAfterCleanRollbackAndClose(){ + var key=RootWorkRejectionKey.create();var r=failure(f->{},(f,s)->Mono.error(new ExpectedRejection()).onErrorResume(ExpectedRejection.class,error->s.reject(key)),1); + assertTrue(r.error.workRejectedBy(key));assertFalse(r.error.workRejectedBy(RootWorkRejectionKey.create()));assertFalse(r.error.workRejectedBy(null)); + assertEquals(RootTransactionException.Stage.ROLLBACK,r.error.stage());assertTrue(r.error.transactionFacts().rollbackAcknowledged());assertFalse(r.error.transactionFacts().commitAcknowledged());assertFalse(r.error.transactionFacts().boundaryCompleted());assertEquals(CleanupFact.COMPLETE,r.error.cleanupFact());assertEquals(1,r.fixture.driver.original.closes); + assertEquals("RootWorkRejectionKey",key.toString()); + } + @Test void defaultScopeImplementationCannotCreateARejectionCertificate(){ + RootTransactionScope unsupported=new RootTransactionScope(){public Duration remaining(){return Duration.ofSeconds(1);}public void narrow(R2dbcOperationBudget cap){}}; + StepVerifier.create(unsupported.reject(RootWorkRejectionKey.create())).expectError(IllegalStateException.class).verify(); + } + @Test void oldConstructorsAndUnclassifiedErrorsNeverAcquireAKey(){ + var key=RootWorkRejectionKey.create();var r=failure(f->{},(f,s)->Mono.error(new ExpectedRejection()),1);assertFalse(r.error.workRejectedBy(key)); + var old=new RootTransactionException(r.error.transactionFacts(),r.error.cleanupFact(),r.error.stage());assertFalse(old.workRejectedBy(key));assertEquals(r.error.getMessage(),old.getMessage());assertNull(old.getCause()); + } + @Test void replacingThePrivateWorkSignalRefusesClassification(){ + var key=RootWorkRejectionKey.create();var r=failure(f->{},(f,s)->s.reject(key).onErrorMap(error->new IllegalStateException("PRIVATE-REPLACEMENT")),1);assertFalse(r.error.workRejectedBy(key)); + } + @Test void anEarlierTypedMarkerCannotClassifyTheLaterOwnedDeadline(){ + var key=RootWorkRejectionKey.create();var marker=new AtomicReference();var r=failure(f->{},(f,s)->Mono.error(new ExpectedRejection()).doOnError(marker::set).onErrorResume(error->Mono.never()),8); + assertInstanceOf(ExpectedRejection.class,marker.get());assertFalse(r.error.workRejectedBy(key));assertTrue(r.error.transactionFacts().rollbackAcknowledged());assertEquals(CleanupFact.COMPLETE,r.error.cleanupFact()); + } + @Test void aRecoveredPrivateSignalCannotClassifyTheLaterOwnedDeadline(){ + var key=RootWorkRejectionKey.create();var r=failure(f->{},(f,s)->s.reject(key).onErrorResume(error->Mono.never()),8);assertFalse(r.error.workRejectedBy(key));assertEquals(0,r.fixture.driver.original.commits); + } + @Test void rollbackFailureAndCloseTimeoutWithholdTheKey(){ + for(boolean rollback:List.of(true,false)){ + var key=RootWorkRejectionKey.create();var r=failure(f->{f.driver.original.failRollback=rollback;f.driver.holdClose=!rollback;},(f,s)->s.reject(key),8); + assertFalse(r.error.workRejectedBy(key));assertEquals(CleanupFact.INCOMPLETE,r.error.cleanupFact());assertEquals(1,r.fixture.driver.original.closes); + } + } + @Test void pendingRollbackWithholdsTheErrorUntilItsAcknowledgment(){ + var key=RootWorkRejectionKey.create();var fixture=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();fixture.set(f);f.driver.holdRollback=true;return f.execute(s->s.reject(key));}) + .expectSubscription().expectNoEvent(Duration.ofMillis(100)).then(()->{assertEquals(1,fixture.get().driver.original.rollbacks);assertEquals(0,fixture.get().driver.original.closes);fixture.get().driver.rollbackGate.tryEmitEmpty();}) + .expectErrorSatisfies(error->assertTrue(assertInstanceOf(RootTransactionException.class,error).workRejectedBy(key))).verify(Duration.ofSeconds(1)); + } + @Test void pendingCloseWithholdsTheCertificateUntilTheOriginalCloseCompletes(){ + var key=RootWorkRejectionKey.create();var fixture=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();fixture.set(f);f.driver.holdClose=true;return f.execute(s->s.reject(key));}) + .expectSubscription().expectNoEvent(Duration.ofMillis(100)).then(()->{assertEquals(1,fixture.get().driver.original.closes);fixture.get().driver.closeGate.tryEmitEmpty();}) + .expectErrorSatisfies(error->assertTrue(assertInstanceOf(RootTransactionException.class,error).workRejectedBy(key))).verify(Duration.ofSeconds(1)); + } + @Test void finalOriginalBudgetExpiryDuringCloseCannotRetainClassification(){ + var key=RootWorkRejectionKey.create();var r=failure(f->f.driver.original.afterClose=()->f.time.wallOffset.addAndGet(6000),(f,s)->s.reject(key),1); + assertFalse(r.error.workRejectedBy(key));assertEquals(CleanupFact.COMPLETE,r.error.cleanupFact());assertTrue(r.error.transactionFacts().rollbackAcknowledged()); + } + @Test void commitAcknowledgmentOrCommitTimeoutCannotBecomeAWorkRejection(){ + for(boolean stall:List.of(false,true)){ + var key=RootWorkRejectionKey.create();var r=failure(f->f.driver.original.stallCommit=stall,(f,s)->stall?Mono.just("value"):TransactionSynchronizationManager.forCurrentTransaction().flatMap(sync->{sync.registerSynchronization(new TransactionSynchronization(){@Override public Mono afterCommit(){return s.reject(key);}});return Mono.just("value");}),8); + assertFalse(r.error.workRejectedBy(key));assertEquals(!stall,r.error.transactionFacts().commitAcknowledged()); + } + } + @Test void aConsumedKeyCannotBeUsedByAnotherInvocation(){ + var key=RootWorkRejectionKey.create();var outcomes=new ArrayList(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();return f.execute(s->s.reject(key)).materialize().concatWith(f.execute(s->s.reject(key)).materialize());}) + .assertNext(signal->{var error=assertInstanceOf(RootTransactionException.class,signal.getThrowable());outcomes.add(error);assertTrue(error.workRejectedBy(key));}) + .assertNext(signal->{var error=assertInstanceOf(RootTransactionException.class,signal.getThrowable());outcomes.add(error);assertFalse(error.workRejectedBy(key));}).verifyComplete();assertEquals(2,outcomes.size()); + } + @Test void oldAndForeignScopesCannotEmitAClassifiedSignal(){ + var first=RootWorkRejectionKey.create();var second=RootWorkRejectionKey.create();var saved=new AtomicReference();var context=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();return f.execute(s->Mono.deferContextual(c->{saved.set(s);context.set(c);return s.reject(first);}));}).expectErrorSatisfies(error->assertTrue(((RootTransactionException)error).workRejectedBy(first))).verify(); + StepVerifier.create(saved.get().reject(second).contextWrite(context.get())).expectError(IllegalStateException.class).verify(); + var r=failure(f->{},(f,s)->saved.get().reject(second),1);assertFalse(r.error.workRejectedBy(second)); + } + @Test void callerCancellationCannotReviveAConsumedKey(){ + var key=RootWorkRejectionKey.create();var fixture=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();fixture.set(f);return f.execute(s->s.reject(key).onErrorResume(error->Mono.never()));}).thenAwait(Duration.ofMillis(100)).thenCancel().verify(Duration.ofSeconds(1)); + assertEquals(1,fixture.get().driver.original.closes);var next=failure(f->{},(f,s)->s.reject(key),1);assertFalse(next.error.workRejectedBy(key)); + } + @Test void swallowingTheRejectionStillCannotCommitBusinessWork(){ + var key=RootWorkRejectionKey.create();var fixture=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();fixture.set(f);return f.execute(s->s.reject(key).onErrorReturn("recovered"));}) + .assertNext(value->{assertTrue(value.transactionFacts().rollbackAcknowledged());assertFalse(value.transactionFacts().commitAcknowledged());assertTrue(value.transactionFacts().boundaryCompleted());}).verifyComplete();assertEquals(0,fixture.get().driver.original.commits); + } + @Test void freshSubscriberKeysDoNotShareClassificationState(){ + var keys=new ArrayList();var errors=new ArrayList(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();return Mono.defer(()->{var key=RootWorkRejectionKey.create();keys.add(key);return f.execute(s->s.reject(key)).materialize();}).repeat(1);}) + .assertNext(signal->errors.add(assertInstanceOf(RootTransactionException.class,signal.getThrowable()))) + .assertNext(signal->errors.add(assertInstanceOf(RootTransactionException.class,signal.getThrowable()))).verifyComplete(); + assertNotSame(keys.getFirst(),keys.getLast());assertTrue(errors.getFirst().workRejectedBy(keys.getFirst()));assertTrue(errors.getLast().workRejectedBy(keys.getLast()));assertFalse(errors.getFirst().workRejectedBy(keys.getLast())); + } + @Test void controlledWorkAndDeadlineWinnersCannotBeConfused(){ + for(long workMillis:List.of(50L,150L)){ + var key=RootWorkRejectionKey.create();var r=failure(f->{},(f,s)->{ + s.narrow(f.time.budget(100));return Mono.delay(Duration.ofMillis(workMillis)).then(Mono.defer(()->s.reject(key))); + },1); + assertEquals(workMillis==50,r.error.workRejectedBy(key));assertTrue(r.error.transactionFacts().rollbackAcknowledged());assertEquals(CleanupFact.COMPLETE,r.error.cleanupFact()); + } + } + @Test void deadlineDuringPendingRollbackStaysUnclassifiedAfterCleanLateCompletion(){ + var key=RootWorkRejectionKey.create();var fixture=new AtomicReference(); + StepVerifier.withVirtualTime(()->{var f=new Fixture();fixture.set(f);f.driver.holdRollback=true;return f.execute(s->{s.narrow(f.time.budget(100));return s.reject(key);});}) + .expectSubscription().expectNoEvent(Duration.ofMillis(150)).then(()->fixture.get().driver.rollbackGate.tryEmitEmpty()) + .expectErrorSatisfies(error->{var root=assertInstanceOf(RootTransactionException.class,error);assertFalse(root.workRejectedBy(key));assertTrue(root.transactionFacts().rollbackAcknowledged());assertFalse(root.transactionFacts().commitAcknowledged());assertFalse(root.transactionFacts().boundaryCompleted());assertEquals(CleanupFact.COMPLETE,root.cleanupFact());}).verify(Duration.ofSeconds(1)); + } +}