package jnpf.flowable.service;
|
|
import com.baomidou.dynamic.datasource.tx.*;
|
import jnpf.exception.WorkFlowException;
|
import jnpf.flowable.mapper.DmsLifecycleEngineRecoveryMapper;
|
import jnpf.flowable.model.flowable.FlowAbleUrl;
|
import lombok.Data;
|
import org.mybatis.spring.SqlSessionTemplate;
|
import org.mybatis.spring.SqlSessionUtils;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import java.util.*;
|
import java.util.function.Supplier;
|
|
/** Durable DMS-only recovery; native business CRUD and ordinary engine calls are unchanged. */
|
@Service
|
public class DmsLifecycleEngineRecovery {
|
private static final Logger LOG=LoggerFactory.getLogger(DmsLifecycleEngineRecovery.class);
|
public static final String CORRELATION_VARIABLE="dmsLifecycleRecoveryId";
|
private final SqlSessionTemplate sessions;
|
private final FlowAbleUrl engine;
|
private final boolean enabled;
|
public DmsLifecycleEngineRecovery(SqlSessionTemplate sessions,FlowAbleUrl engine,
|
@Value("${dms.business-occupancy.enabled:false}") boolean enabled) {
|
this.sessions=sessions;this.engine=engine;this.enabled=enabled;
|
}
|
private DmsLifecycleEngineRecoveryMapper mapper() {return sessions.getMapper(DmsLifecycleEngineRecoveryMapper.class);}
|
|
/** A log is committed BEFORE starting the engine; the native transaction holds its recovery gate. */
|
public Started start(String tenant,String task,String deployment,Map<String,Object> variables) throws WorkFlowException {
|
if(!enabled || task==null || task.isBlank()) throw new IllegalStateException("DMS 引擎恢复未启用或任务标识缺失");
|
if(variables.containsKey(CORRELATION_VARIABLE)) throw new IllegalArgumentException("DMS 引擎恢复变量冲突");
|
var row=new Record();row.setId(UUID.randomUUID().toString());row.setTenantId(tenant==null || tenant.isBlank()?"0":tenant);row.setTaskId(task);
|
holdNativeGate(row.getId());
|
independentChecked(()->{if(mapper().insert(row)!=1)throw new IllegalStateException("DMS 引擎恢复记录未保存");return null;});
|
var actual=new HashMap<>(variables);actual.put(CORRELATION_VARIABLE,row.getId());
|
String instance=engine.startInstance(deployment,actual);
|
if(instance==null || instance.isBlank()) throw new WorkFlowException("DMS 引擎启动结果不明确，需按恢复记录核对");
|
row.setInstanceId(instance);
|
var ticket=DmsLifecycleEngineCompensation.register(instance,()->{
|
independentChecked(()->{
|
// Even a commit-response failure must not delete an instance already referenced by committed data.
|
if(mapper().committed(row)) {mapper().finish(row.getId(),"KEPT");return null;}
|
try {engine.deleteDmsInstanceStrict(instance,reason(row.getId()));}
|
catch(WorkFlowException failure) {throw new IllegalStateException(failure);}
|
mapper().finish(row.getId(),"DELETED");return null;
|
});
|
});
|
try {
|
independentChecked(()->{if(mapper().attach(row.getId(),instance)!=1)throw new IllegalStateException("DMS 引擎实例恢复关联未保存");return null;});
|
} catch(RuntimeException failure) {ticket.compensate(failure);throw failure;}
|
return new Started(instance,ticket);
|
}
|
private void holdNativeGate(String id) {
|
withVerifiedConnection(()->{
|
if(!mapper().tryLock(id)) throw new IllegalStateException("DMS 引擎恢复门闩冲突");
|
return null;
|
});
|
}
|
private <T> T independentChecked(Supplier<T> action) {
|
return independent(()->withVerifiedConnection(action));
|
}
|
private <T> T withVerifiedConnection(Supplier<T> action) {
|
String xid=TransactionContext.getXID();
|
if(xid==null || TransactionSynchronizationManager.isActualTransactionActive())
|
throw new IllegalStateException("DMS 引擎启动必须属于原生动态事务");
|
var session=SqlSessionUtils.getSqlSession(sessions.getSqlSessionFactory(),sessions.getExecutorType(),sessions.getPersistenceExceptionTranslator());
|
try {
|
var connection=session.getConnection();
|
if(!(connection instanceof ConnectionProxy proxy) || ConnectionFactory.getConnection(xid,proxy.getDs())!=connection
|
|| connection.getAutoCommit() || connection.isReadOnly()
|
|| connection.getTransactionIsolation()!=java.sql.Connection.TRANSACTION_READ_COMMITTED)
|
throw new IllegalStateException("DMS 引擎恢复门闩未绑定原生事务连接");
|
return action.get();
|
} catch(java.sql.SQLException failure) {throw new IllegalStateException(failure);}
|
finally {SqlSessionUtils.closeSqlSession(session,sessions.getSqlSessionFactory());}
|
}
|
/** REQUIRES_NEW suspends the actual dynamic transaction; Spring propagation is not substituted. */
|
static <T> T independent(Supplier<T> action) {
|
if(TransactionSynchronizationManager.isActualTransactionActive()) throw new IllegalStateException("不能混用DMS恢复事务管理器");
|
// Version 4.3.1 suspends only XID, not its thread-local synchronization list.
|
// Isolate callbacks here so an independent journal transaction cannot consume native callbacks.
|
String outerXid=TransactionContext.getXID();
|
var outerCallbacks=TransactionContext.getSynchronizations();
|
TransactionContext.removeSynchronizations();
|
var info=new TransactionalInfo();info.setPropagation(DsPropagation.REQUIRES_NEW);
|
info.setRollbackFor(new Class[]{Exception.class});info.setNoRollbackFor(new Class[]{});
|
try {
|
@SuppressWarnings("unchecked") T result=(T)new TransactionalTemplate().execute(new TransactionalExecutor() {
|
public Object execute() {return action.get();}
|
public TransactionalInfo getTransactionInfo() {return info;}
|
});return result;
|
} catch(RuntimeException | Error failure) {throw failure;}
|
catch(Throwable failure) {throw new IllegalStateException("DMS 独立恢复事务失败",failure);}
|
finally {
|
TransactionContext.removeSynchronizations();
|
// afterCompletion runs after LocalTxUtil has already removed XID.
|
// Its callback snapshot is still being iterated, but no live transaction can be re-registered.
|
if(outerXid!=null && outerXid.equals(TransactionContext.getXID()))
|
outerCallbacks.forEach(TransactionContext::registerSynchronization);
|
}
|
}
|
@Scheduled(cron="${dms.business-occupancy.engine-recovery-cron:0 * * * * ?}")
|
public void recoverPending() {
|
if(!enabled) return;
|
for(var candidate:mapper().pending(100)) {
|
try {recover(candidate.getId());}
|
catch(RuntimeException failure) {LOG.error("DMS engine recovery scan failed, recoveryId={}",candidate.getId(),failure);}
|
}
|
}
|
public void recover(String id) {
|
independentChecked(()->{
|
if(!mapper().tryLock(id)) return null; // A live native transaction still owns this instance.
|
var row=mapper().get(id);
|
if(row==null || !"PENDING".equals(row.getStatus())) return null;
|
if(row.getInstanceId()==null || row.getInstanceId().isBlank()) {
|
mapper().finish(id,"NEEDS_REVIEW");return null; // Never guess which remote instance to delete.
|
}
|
if(mapper().committed(row)) {mapper().finish(id,"KEPT");return null;}
|
try {
|
engine.deleteDmsInstanceStrict(row.getInstanceId(),reason(id));
|
mapper().finish(id,"DELETED");
|
} catch(Exception failure) {
|
String type=failure.getClass().getName();
|
mapper().failed(id,type.substring(0,Math.min(200,type.length())));
|
LOG.warn("DMS engine deletion unconfirmed, recoveryId={}",id);
|
}
|
return null;
|
});
|
}
|
private static String reason(String id) {return "dmsLifecycleRollback:"+id;}
|
public record Started(String instanceId,DmsLifecycleEngineCompensation compensation) { }
|
@Data public static class Record {
|
private String id;
|
private String tenantId;
|
private String taskId;
|
private String instanceId;
|
private String status;
|
}
|
}
|
