package jnpf.flowable.service;
|
|
import com.baomidou.dynamic.datasource.tx.*;
|
import com.alibaba.fastjson.JSONObject;
|
import jnpf.config.ConfigValueUtil;
|
import jnpf.exception.WorkFlowException;
|
import jnpf.flowable.mapper.DmsLifecycleEngineRecoveryMapper;
|
import jnpf.flowable.model.flowable.FlowAbleUrl;
|
import jnpf.util.UserProvider;
|
import jnpf.util.wxutil.HttpUtil;
|
import org.junit.After;
|
import org.junit.Test;
|
import org.mybatis.spring.SqlSessionTemplate;
|
import java.util.concurrent.atomic.AtomicInteger;
|
import static org.junit.Assert.*;
|
import static org.mockito.Mockito.*;
|
|
public class DmsLifecycleEngineRecoveryTest {
|
@After public void clean(){TransactionContext.remove();TransactionContext.removeSynchronizations();}
|
private static ConnectionProxy enlistConnection() {
|
String xid=TransactionContext.getXID();
|
var existing=ConnectionFactory.getConnection(xid,"master");
|
if(existing!=null) return existing;
|
var physical=mock(java.sql.Connection.class);
|
try {when(physical.getTransactionIsolation()).thenReturn(java.sql.Connection.TRANSACTION_READ_COMMITTED);}
|
catch(java.sql.SQLException impossible){throw new AssertionError(impossible);}
|
var proxy=new ConnectionProxy(physical,"master");
|
ConnectionFactory.putConnection(xid,"master",proxy);return proxy;
|
}
|
private org.mockito.MockedStatic<org.mybatis.spring.SqlSessionUtils> sessions(SqlSessionTemplate template,DmsLifecycleEngineRecoveryMapper mapper) {
|
var factory=mock(org.apache.ibatis.session.SqlSessionFactory.class);
|
var session=mock(org.apache.ibatis.session.SqlSession.class);
|
when(template.getSqlSessionFactory()).thenReturn(factory);
|
when(template.getMapper(DmsLifecycleEngineRecoveryMapper.class)).thenReturn(mapper);
|
when(session.getConnection()).thenAnswer(invocation->enlistConnection());
|
var utils=mockStatic(org.mybatis.spring.SqlSessionUtils.class);
|
utils.when(()->org.mybatis.spring.SqlSessionUtils.getSqlSession(eq(factory),any(),any())).thenReturn(session);
|
return utils;
|
}
|
@Test public void independentRecoveryCommitSurvivesOuterDynamicRollback() throws Throwable {
|
var committed=new AtomicInteger();var outerRollback=new AtomicInteger();
|
var info=new TransactionalInfo();info.setPropagation(DsPropagation.REQUIRED);info.setRollbackFor(new Class[]{Exception.class});info.setNoRollbackFor(new Class[]{});
|
try {new TransactionalTemplate().execute(new TransactionalExecutor(){
|
public TransactionalInfo getTransactionInfo(){return info;}
|
public Object execute(){
|
enlistConnection();
|
String xid=TransactionContext.getXID();
|
TransactionContext.registerSynchronization(new org.springframework.transaction.support.TransactionSynchronization(){
|
public void afterCompletion(int status){if(status==STATUS_ROLLED_BACK)outerRollback.incrementAndGet();}
|
});
|
DmsLifecycleEngineRecovery.independent(()->{
|
enlistConnection();
|
assertNotEquals(xid,TransactionContext.getXID());
|
TransactionContext.registerSynchronization(new org.springframework.transaction.support.TransactionSynchronization(){
|
public void afterCompletion(int status){if(status==STATUS_COMMITTED)committed.incrementAndGet();}
|
});return null;
|
});assertEquals(xid,TransactionContext.getXID());throw new IllegalStateException("native failure");
|
}
|
});fail();}catch(IllegalStateException expected){}
|
assertEquals(1,committed.get());assertEquals(1,outerRollback.get());
|
}
|
@Test public void startJournalsBeforeRemoteCallAndRetainsFailureBeyondRollback() throws Throwable {
|
var sessions=mock(SqlSessionTemplate.class);var mapper=mock(DmsLifecycleEngineRecoveryMapper.class);
|
var engine=mock(FlowAbleUrl.class);
|
when(mapper.tryLock(anyString())).thenReturn(true);when(mapper.insert(any())).thenReturn(1);when(mapper.attach(anyString(),eq("engine"))).thenReturn(1);
|
when(engine.startInstance(eq("deployment"),anyMap())).thenReturn("engine");
|
doThrow(new WorkFlowException("offline")).when(engine).deleteDmsInstanceStrict(eq("engine"),anyString());
|
var info=new TransactionalInfo();info.setPropagation(DsPropagation.REQUIRED);info.setRollbackFor(new Class[]{Exception.class});info.setNoRollbackFor(new Class[]{});
|
try(var utils=sessions(sessions,mapper)) {
|
try {new TransactionalTemplate().execute(new TransactionalExecutor(){
|
public TransactionalInfo getTransactionInfo(){return info;}
|
public Object execute(){
|
enlistConnection();
|
try {
|
var started=new DmsLifecycleEngineRecovery(sessions,engine,true).start("tenant","task","deployment",java.util.Map.of());
|
var failure=new IllegalStateException("later native failure");started.compensation().compensate(failure);throw failure;
|
} catch(WorkFlowException failure){throw new IllegalStateException(failure);}
|
}
|
});fail();}catch(IllegalStateException expected){}
|
}
|
var order=inOrder(mapper,engine);order.verify(mapper).insert(any());order.verify(engine).startInstance(eq("deployment"),argThat(v->v.containsKey(DmsLifecycleEngineRecovery.CORRELATION_VARIABLE)));
|
order.verify(mapper).attach(anyString(),eq("engine"));
|
verify(engine,times(2)).deleteDmsInstanceStrict(eq("engine"),anyString());
|
verify(mapper,never()).finish(anyString(),anyString());
|
assertTrue(TransactionContext.getSynchronizations().isEmpty());
|
}
|
private void recoveryCase(boolean lock,String instance,boolean committed,boolean deletionFails,String expected) throws Exception {
|
var sessions=mock(SqlSessionTemplate.class);var mapper=mock(DmsLifecycleEngineRecoveryMapper.class);var engine=mock(FlowAbleUrl.class);
|
when(sessions.getMapper(DmsLifecycleEngineRecoveryMapper.class)).thenReturn(mapper);
|
when(mapper.tryLock("r")).thenReturn(lock);
|
var row=new DmsLifecycleEngineRecovery.Record();row.setId("r");row.setTenantId("t");row.setTaskId("task");row.setInstanceId(instance);row.setStatus("PENDING");
|
when(mapper.get("r")).thenReturn(row);when(mapper.committed(row)).thenReturn(committed);
|
if(deletionFails)doThrow(new WorkFlowException("no acknowledgement")).when(engine).deleteDmsInstanceStrict("engine","dmsLifecycleRollback:r");
|
try(var utils=sessions(sessions,mapper)){new DmsLifecycleEngineRecovery(sessions,engine,true).recover("r");}
|
if(!lock){verify(mapper,never()).get(anyString());verifyNoInteractions(engine);}
|
else if(deletionFails){verify(mapper).failed(eq("r"),anyString());verify(mapper,never()).finish(anyString(),anyString());}
|
else {verify(mapper).finish("r",expected);if(!"DELETED".equals(expected))verifyNoInteractions(engine);}
|
}
|
@Test public void activeNativeTransactionCannotBeRecovered() throws Exception {recoveryCase(false,"engine",false,false,null);}
|
@Test public void committedNativeTaskIsNeverDeleted() throws Exception {recoveryCase(true,"engine",true,false,"KEPT");}
|
@Test public void unknownStartResultRequiresEvidence() throws Exception {recoveryCase(true,null,false,false,"NEEDS_REVIEW");}
|
@Test public void confirmedDeletionCompletesRecovery() throws Exception {recoveryCase(true,"engine",false,false,"DELETED");}
|
@Test public void deletionFailureRetainsPendingRecovery() throws Exception {recoveryCase(true,"engine",false,true,null);}
|
@Test public void strictClientRejectsFalseEmptyAndNullAcknowledgements() throws Exception {
|
var config=mock(ConfigValueUtil.class);when(config.getFlowDomain()).thenReturn("http://review.invalid");
|
var client=new FlowAbleUrl(config);
|
try(var context=mockStatic(jnpf.util.context.SpringContext.class);var user=mockStatic(UserProvider.class);var http=mockStatic(HttpUtil.class)){
|
user.when(UserProvider::getToken).thenReturn("test-only");
|
for(JSONObject response:new JSONObject[]{null,new JSONObject(),JSONObject.parseObject("{\"success\":false}")}){
|
http.when(()->HttpUtil.httpRequest(anyString(),eq("DELETE"),isNull(),eq("test-only"))).thenReturn(response);
|
http.when(()->HttpUtil.httpRequest(anyString(),eq("GET"),isNull(),eq("test-only"))).thenReturn(new JSONObject());
|
try{client.deleteDmsInstanceStrict("engine","reason");fail();}catch(WorkFlowException expected){}
|
}
|
}
|
}
|
@Test public void lostDeletionResponseNeedsExactHistoricReceipt() throws Exception {
|
var config=mock(ConfigValueUtil.class);when(config.getFlowDomain()).thenReturn("http://review.invalid");var client=new FlowAbleUrl(config);
|
try(var context=mockStatic(jnpf.util.context.SpringContext.class);var user=mockStatic(UserProvider.class);var http=mockStatic(HttpUtil.class)){
|
user.when(UserProvider::getToken).thenReturn("test-only");
|
http.when(()->HttpUtil.httpRequest(anyString(),eq("DELETE"),isNull(),eq("test-only"))).thenReturn(JSONObject.parseObject("{\"success\":false}"));
|
for(String reason:new String[]{"different","reason"}){
|
var history=JSONObject.parseObject("{\"success\":true,\"data\":{\"instanceId\":\"engine\",\"endTime\":\"2026-10-15T12:00:00\"}}");history.getJSONObject("data").put("deleteReason",reason);
|
http.when(()->HttpUtil.httpRequest(anyString(),eq("GET"),isNull(),eq("test-only"))).thenReturn(history);
|
if("reason".equals(reason))client.deleteDmsInstanceStrict("engine","reason");
|
else try{client.deleteDmsInstanceStrict("engine","reason");fail();}catch(WorkFlowException expected){}
|
}
|
}
|
}
|
}
|
