fix(shadow): repair UTC sampling and persist v3 diagnostics
This commit is contained in:
@@ -9,6 +9,7 @@ import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.sql.SQLException;
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
@@ -346,9 +347,32 @@ public class ShadowObservationGateway {
|
||||
long elapsedMicros = startedNanos == 0L ? 0L
|
||||
: TimeUnit.NANOSECONDS.toMicros(Math.max(0L, System.nanoTime() - startedNanos));
|
||||
LOG.warn("shadow_failure requestId={} warehouse={} snapshotGeneration={} stage={} "
|
||||
+ "exceptionClass={} elapsedMicros={}",
|
||||
+ "exceptionClass={} elapsedMicros={} {}",
|
||||
requestId, warehouseId, generation, stage,
|
||||
failure.getClass().getName(), elapsedMicros);
|
||||
failure.getClass().getName(), elapsedMicros, diagnosticMetadata(failure));
|
||||
}
|
||||
|
||||
private static String diagnosticMetadata(Throwable failure) {
|
||||
Throwable cause = failure;
|
||||
SQLException sqlFailure = null;
|
||||
// Bounded traversal; never log exception messages, SQL text or wrapped business payloads.
|
||||
for (int depth = 0; depth < 8; depth++) {
|
||||
if (cause instanceof SQLException sql) {
|
||||
sqlFailure = sql;
|
||||
}
|
||||
if (cause.getCause() == null || cause.getCause() == cause || depth == 7) {
|
||||
break;
|
||||
}
|
||||
cause = cause.getCause();
|
||||
}
|
||||
String sqlState = sqlFailure == null ? null : sqlFailure.getSQLState();
|
||||
StackTraceElement[] trace = cause.getStackTrace();
|
||||
String source = trace.length == 0 ? "-"
|
||||
: trace[0].getClassName() + '.' + trace[0].getMethodName() + ':' + trace[0].getLineNumber();
|
||||
return "causeClass=" + cause.getClass().getName()
|
||||
+ " sqlState=" + (sqlState != null && sqlState.matches("[A-Za-z0-9]{5}") ? sqlState : "-")
|
||||
+ " sqlCode=" + (sqlFailure == null ? 0 : sqlFailure.getErrorCode())
|
||||
+ " causeAt=" + source;
|
||||
}
|
||||
|
||||
private String failureDescription(ShadowFailureStage stage, Throwable failure) {
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
package com.ruoyi.algorithm.shadow.mapper;
|
||||
|
||||
import org.apache.ibatis.type.BaseTypeHandler;
|
||||
import org.apache.ibatis.type.JdbcType;
|
||||
|
||||
import java.sql.CallableStatement;
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Timestamp;
|
||||
import java.time.Instant;
|
||||
import java.util.Calendar;
|
||||
import java.util.TimeZone;
|
||||
|
||||
/** Explicitly referenced only by shadow mappings whose datetime2 values are UTC. */
|
||||
public final class ShadowUtcInstantTypeHandler extends BaseTypeHandler<Instant> {
|
||||
@Override
|
||||
public void setNonNullParameter(PreparedStatement statement, int index, Instant value, JdbcType jdbcType)
|
||||
throws SQLException {
|
||||
statement.setTimestamp(index, Timestamp.from(value), utcCalendar());
|
||||
}
|
||||
|
||||
@Override
|
||||
public Instant getNullableResult(ResultSet results, String column) throws SQLException {
|
||||
return instant(results.getTimestamp(column, utcCalendar()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Instant getNullableResult(ResultSet results, int column) throws SQLException {
|
||||
return instant(results.getTimestamp(column, utcCalendar()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Instant getNullableResult(CallableStatement statement, int column) throws SQLException {
|
||||
return instant(statement.getTimestamp(column, utcCalendar()));
|
||||
}
|
||||
|
||||
private static Instant instant(Timestamp timestamp) {
|
||||
return timestamp == null ? null : timestamp.toInstant();
|
||||
}
|
||||
|
||||
private static Calendar utcCalendar() {
|
||||
// Calendar is mutable and may be modified by a JDBC driver; never share it across calls.
|
||||
return Calendar.getInstance(TimeZone.getTimeZone("UTC"));
|
||||
}
|
||||
}
|
||||
@@ -71,7 +71,28 @@
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- 系统模块日志级别控制 -->
|
||||
<!-- Shadow diagnostics only; business appenders and their filters remain unchanged. -->
|
||||
<appender name="file_shadow" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/shadow-diagnostics.log</file>
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
|
||||
<fileNamePattern>${log.path}/shadow-diagnostics.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
|
||||
<maxFileSize>10MB</maxFileSize>
|
||||
<maxHistory>14</maxHistory>
|
||||
<totalSizeCap>100MB</totalSizeCap>
|
||||
</rollingPolicy>
|
||||
<encoder>
|
||||
<charset>UTF-8</charset>
|
||||
<pattern>%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} [%thread] %-5level %logger - %msg%n</pattern>
|
||||
</encoder>
|
||||
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
|
||||
<level>WARN</level>
|
||||
</filter>
|
||||
</appender>
|
||||
<logger name="com.ruoyi.algorithm.shadow">
|
||||
<appender-ref ref="file_shadow"/>
|
||||
</logger>
|
||||
|
||||
<!-- 系统模块日志级别控制 -->
|
||||
<logger name="com.ruoyi" level="info" />
|
||||
<!-- Spring日志级别控制 -->
|
||||
<logger name="org.springframework" level="warn" />
|
||||
@@ -90,4 +111,4 @@
|
||||
<logger name="sys-user" level="info">
|
||||
<appender-ref ref="sys-user"/>
|
||||
</logger>
|
||||
</configuration>
|
||||
</configuration>
|
||||
|
||||
@@ -89,8 +89,16 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
@instance_id = #{instanceId};
|
||||
</update>
|
||||
|
||||
<select id="selectPublishedSnapshot"
|
||||
resultType="com.ruoyi.algorithm.shadow.ShadowSnapshotDescriptor">
|
||||
<resultMap id="publishedSnapshot" type="com.ruoyi.algorithm.shadow.ShadowSnapshotDescriptor">
|
||||
<result property="snapshotGeneration" column="snapshotGeneration"/>
|
||||
<result property="warehouseId" column="warehouseId"/>
|
||||
<result property="snapshotStatus" column="snapshotStatus"/>
|
||||
<result property="synchronizedAt" column="synchronizedAt"
|
||||
typeHandler="com.ruoyi.algorithm.shadow.mapper.ShadowUtcInstantTypeHandler"/>
|
||||
<result property="sourceSignature" column="sourceSignature"/>
|
||||
</resultMap>
|
||||
|
||||
<select id="selectPublishedSnapshot" resultMap="publishedSnapshot">
|
||||
select state.snapshot_generation as snapshotGeneration,
|
||||
state.warehouse_id as warehouseId,
|
||||
state.snapshot_status as snapshotStatus,
|
||||
@@ -109,7 +117,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
where state.warehouse_id = #{warehouseId}
|
||||
and state.snapshot_status = 'PUBLISHED'
|
||||
and state.snapshot_generation = #{inputVersion}
|
||||
and state.synchronized_at = #{snapshotSynchronizedAt}
|
||||
and state.synchronized_at = #{snapshotSynchronizedAt, jdbcType=TIMESTAMP, typeHandler=com.ruoyi.algorithm.shadow.mapper.ShadowUtcInstantTypeHandler}
|
||||
and convert(varchar(64), state.source_signature, 2) = #{sourceSignature}
|
||||
</insert>
|
||||
|
||||
@@ -126,12 +134,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
metric_total_micros = target.metric_total_micros + #{metricTotalMicros},
|
||||
metric_max_micros = case when target.metric_max_micros > #{metricMaxMicros}
|
||||
then target.metric_max_micros else #{metricMaxMicros} end,
|
||||
flushed_at = #{flushedAt}
|
||||
flushed_at = #{flushedAt, jdbcType=TIMESTAMP, typeHandler=com.ruoyi.algorithm.shadow.mapper.ShadowUtcInstantTypeHandler}
|
||||
when not matched then insert (
|
||||
warehouse_id, metric_name, bucket_upper_micros,
|
||||
metric_count, metric_total_micros, metric_max_micros, flushed_at)
|
||||
values (
|
||||
source.warehouse_id, source.metric_name, source.bucket_upper_micros,
|
||||
#{metricCount}, #{metricTotalMicros}, #{metricMaxMicros}, #{flushedAt});
|
||||
#{metricCount}, #{metricTotalMicros}, #{metricMaxMicros}, #{flushedAt, jdbcType=TIMESTAMP, typeHandler=com.ruoyi.algorithm.shadow.mapper.ShadowUtcInstantTypeHandler});
|
||||
</update>
|
||||
</mapper>
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
package com.ruoyi.algorithm.shadow;
|
||||
|
||||
import ch.qos.logback.classic.LoggerContext;
|
||||
import ch.qos.logback.classic.joran.JoranConfigurator;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
import javax.xml.parsers.DocumentBuilderFactory;
|
||||
import javax.xml.transform.TransformerFactory;
|
||||
import javax.xml.transform.dom.DOMSource;
|
||||
import javax.xml.transform.stream.StreamResult;
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class ShadowLogRoutingTest {
|
||||
@TempDir
|
||||
Path logs;
|
||||
|
||||
@Test
|
||||
void shadowWarningsArePersistedWithoutChangingBusinessLogRouting() throws Exception {
|
||||
var factory = DocumentBuilderFactory.newInstance();
|
||||
factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true);
|
||||
var document = factory.newDocumentBuilder().parse(getClass().getResourceAsStream("/logback.xml"));
|
||||
var properties = document.getElementsByTagName("property");
|
||||
for (int index = 0; index < properties.getLength(); index++) {
|
||||
Element property = (Element) properties.item(index);
|
||||
if ("log.path".equals(property.getAttribute("name"))) {
|
||||
property.setAttribute("value", logs.toString().replace('\\', '/'));
|
||||
}
|
||||
}
|
||||
var xml = new ByteArrayOutputStream();
|
||||
TransformerFactory.newInstance().newTransformer().transform(new DOMSource(document), new StreamResult(xml));
|
||||
LoggerContext context = new LoggerContext();
|
||||
context.setMDCAdapter(new ch.qos.logback.classic.util.LogbackMDCAdapter());
|
||||
try {
|
||||
JoranConfigurator configurator = new JoranConfigurator();
|
||||
configurator.setContext(context);
|
||||
configurator.doConfigure(new ByteArrayInputStream(xml.toByteArray()));
|
||||
var shadow = context.getLogger(ShadowObservationGateway.class);
|
||||
shadow.warn("shadow_failure stage=REFRESH exceptionClass=java.sql.SQLException");
|
||||
shadow.error("shadow_error_preserved");
|
||||
shadow.info("shadow_info_not_a_warning");
|
||||
context.getLogger("com.ruoyi.business.Example").warn("business_warn_not_shadow");
|
||||
context.getLogger("com.ruoyi.business.Example").error("business_error_preserved");
|
||||
assertThat(context.getStatusManager().getCopyOfStatusList())
|
||||
.noneMatch(status -> status.getLevel() == ch.qos.logback.core.status.Status.ERROR);
|
||||
} finally {
|
||||
context.stop();
|
||||
}
|
||||
Path shadowLog = logs.resolve("shadow-diagnostics.log");
|
||||
assertThat(shadowLog).isRegularFile();
|
||||
assertThat(Files.readString(shadowLog)).contains("shadow_failure", "stage=REFRESH", "shadow_error_preserved")
|
||||
.containsPattern("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}")
|
||||
.doesNotContain("shadow_info_not_a_warning", "business_warn_not_shadow", "business_error_preserved");
|
||||
assertThat(Files.readString(logs.resolve("sys-error.log")))
|
||||
.contains("business_error_preserved", "shadow_error_preserved").doesNotContain("shadow_failure");
|
||||
assertThat(Files.readString(logs.resolve("sys-info.log"))).contains("shadow_info_not_a_warning");
|
||||
}
|
||||
}
|
||||
@@ -214,6 +214,48 @@ class ShadowObservationGatewayTest {
|
||||
verify(engine, never()).compareInbound(any(), anyString(), anyString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void workerFailureLogsSqlCauseWithoutExceptionMessagesOrPayloads() throws Exception {
|
||||
var logger = (ch.qos.logback.classic.Logger) org.slf4j.LoggerFactory.getLogger(ShadowObservationGateway.class);
|
||||
var logged = new java.util.concurrent.atomic.AtomicReference<String>();
|
||||
var logThread = new java.util.concurrent.atomic.AtomicReference<String>();
|
||||
CountDownLatch warning = new CountDownLatch(1);
|
||||
var appender = new ch.qos.logback.core.AppenderBase<ch.qos.logback.classic.spi.ILoggingEvent>() {
|
||||
@Override
|
||||
protected void append(ch.qos.logback.classic.spi.ILoggingEvent event) {
|
||||
logged.set(event.getFormattedMessage());
|
||||
logThread.set(Thread.currentThread().getName());
|
||||
warning.countDown();
|
||||
}
|
||||
};
|
||||
appender.setContext(logger.getLoggerContext());
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
try {
|
||||
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
|
||||
var sqlFailure = new java.sql.SQLException("password=secret SQL payload", "40001", 1205);
|
||||
sqlFailure.setStackTrace(new StackTraceElement[]{
|
||||
new StackTraceElement("com.microsoft.sqlserver.jdbc.SQLServerException", "makeFromDatabaseError", "SQLServerException.java", 270)});
|
||||
when(engine.planInbound(SNAPSHOT, "TASK-LOG", "TRAY-LOG"))
|
||||
.thenThrow(new IllegalStateException("private wrapper message", sqlFailure));
|
||||
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
|
||||
stubSuccessfulRecovery(engine, mapper, 43L);
|
||||
ShadowObservationGateway gateway = gateway(engine, mapper, coordinatorWithSnapshot(),
|
||||
new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { });
|
||||
InboundShadowCompletion completion = gateway.observeInbound("TASK-LOG", "TRAY-LOG");
|
||||
completion.complete("10", "SPACE-LOG");
|
||||
|
||||
assertThat(warning.await(2, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(logged.get()).contains("shadow_failure", "stage=PLAN", "java.sql.SQLException",
|
||||
"sqlState=40001", "sqlCode=1205", "makeFromDatabaseError")
|
||||
.doesNotContain("password", "secret", "SQL payload", "private wrapper message");
|
||||
assertThat(logThread.get()).isNotEqualTo(Thread.currentThread().getName());
|
||||
} finally {
|
||||
logger.detachAppender(appender);
|
||||
appender.stop();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void comparisonFailureIsWorkerOnlyAndRecordedByFixedStage() throws Exception {
|
||||
ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class);
|
||||
|
||||
@@ -337,8 +337,8 @@ class ShadowPlanningEngineTest {
|
||||
"from alg.shadow_snapshot_state state with (updlock, holdlock)",
|
||||
"state.snapshot_status = 'published'",
|
||||
"state.snapshot_generation = #{inputversion}",
|
||||
"state.synchronized_at = #{snapshotsynchronizedat}",
|
||||
"convert(varchar(64), state.source_signature, 2) = #{sourcesignature}");
|
||||
"convert(varchar(64), state.source_signature, 2) = #{sourcesignature}")
|
||||
.containsPattern("state\\.synchronized_at = #\\{snapshotsynchronizedat(?:,[^}]+)?}");
|
||||
assertThat(statements.get("selectPublishedSnapshot"))
|
||||
.contains("alg.shadow_snapshot_state", "convert(varchar(64),",
|
||||
"source_signature", ", 2)");
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
package com.ruoyi.algorithm.shadow;
|
||||
|
||||
import org.apache.ibatis.builder.xml.XMLMapperBuilder;
|
||||
import org.apache.ibatis.mapping.MappedStatement;
|
||||
import org.apache.ibatis.scripting.defaults.DefaultParameterHandler;
|
||||
import org.apache.ibatis.session.Configuration;
|
||||
import org.apache.ibatis.type.InstantTypeHandler;
|
||||
import org.apache.ibatis.type.TypeHandler;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.parallel.ResourceLock;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
|
||||
import java.io.InputStream;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.sql.CallableStatement;
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.Timestamp;
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.ZoneOffset;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Calendar;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.TimeZone;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@ResourceLock("default-time-zone")
|
||||
class ShadowUtcMappingTest {
|
||||
private static final String NAMESPACE = "com.ruoyi.algorithm.shadow.mapper.ShadowInputMapper.";
|
||||
private static final Instant SNAPSHOT_TIME = Instant.parse("2026-09-07T01:31:00.1623685Z");
|
||||
private static final LocalDateTime SQL_TIME = LocalDateTime.parse("2026-09-07T01:31:00.1623685");
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(strings = {"UTC", "Asia/Shanghai"})
|
||||
void publishedUtcDatetime2RemainsFreshAndRetainsPrecision(String zone) throws Exception {
|
||||
TimeZone previous = TimeZone.getDefault();
|
||||
try {
|
||||
TimeZone.setDefault(TimeZone.getTimeZone(zone));
|
||||
TypeHandler<Instant> handler = snapshotHandler(configuration());
|
||||
ResultSet row = (ResultSet) Proxy.newProxyInstance(getClass().getClassLoader(),
|
||||
new Class<?>[]{ResultSet.class}, (proxy, method, args) -> {
|
||||
if (method.getName().equals("getTimestamp")) {
|
||||
return driverTimestamp(args.length == 2 ? (Calendar) args[1] : null);
|
||||
}
|
||||
throw new UnsupportedOperationException(method.getName());
|
||||
});
|
||||
Instant decoded = handler.getResult(row, "synchronizedAt");
|
||||
assertThat(decoded).isEqualTo(SNAPSHOT_TIME);
|
||||
assertThat(handler.getResult(row, 1)).isEqualTo(SNAPSHOT_TIME);
|
||||
CallableStatement call = (CallableStatement) Proxy.newProxyInstance(getClass().getClassLoader(),
|
||||
new Class<?>[]{CallableStatement.class}, (proxy, method, args) -> {
|
||||
if (method.getName().equals("getTimestamp")) {
|
||||
return driverTimestamp(args.length == 2 ? (Calendar) args[1] : null);
|
||||
}
|
||||
throw new UnsupportedOperationException(method.getName());
|
||||
});
|
||||
assertThat(handler.getResult(call, 1)).isEqualTo(SNAPSHOT_TIME);
|
||||
Instant now = Instant.parse("2026-09-07T01:31:01Z");
|
||||
assertThat(admission(decoded, now).accepted()).isTrue();
|
||||
assertThat(admission(decoded, SNAPSHOT_TIME.plusSeconds(61)).skipReason())
|
||||
.isEqualTo(ShadowSkipReason.STALE);
|
||||
} finally {
|
||||
TimeZone.setDefault(previous);
|
||||
}
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(strings = {"UTC", "Asia/Shanghai"})
|
||||
void comparisonAndBothMetricTimestampBindingsWriteUtc(String zone) throws Exception {
|
||||
TimeZone previous = TimeZone.getDefault();
|
||||
try {
|
||||
TimeZone.setDefault(TimeZone.getTimeZone(zone));
|
||||
Configuration configuration = configuration();
|
||||
Map<String, Object> values = new HashMap<>();
|
||||
values.put("snapshotSynchronizedAt", SNAPSHOT_TIME);
|
||||
values.put("flushedAt", SNAPSHOT_TIME);
|
||||
assertThat(boundTimestamps(configuration, "insertComparison", values))
|
||||
.containsExactly(SQL_TIME);
|
||||
assertThat(boundTimestamps(configuration, "upsertRuntimeMetric", values))
|
||||
.containsExactly(SQL_TIME, SQL_TIME);
|
||||
assertThat(configuration.getTypeHandlerRegistry().getTypeHandler(Instant.class))
|
||||
.isExactlyInstanceOf(InstantTypeHandler.class);
|
||||
} finally {
|
||||
TimeZone.setDefault(previous);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void nullableTimestampResultsRemainNull() throws Exception {
|
||||
TypeHandler<Instant> handler = snapshotHandler(configuration());
|
||||
ResultSet row = (ResultSet) Proxy.newProxyInstance(getClass().getClassLoader(),
|
||||
new Class<?>[]{ResultSet.class}, (proxy, method, args) -> null);
|
||||
CallableStatement call = (CallableStatement) Proxy.newProxyInstance(getClass().getClassLoader(),
|
||||
new Class<?>[]{CallableStatement.class}, (proxy, method, args) -> null);
|
||||
assertThat(handler.getResult(row, "synchronizedAt")).isNull();
|
||||
assertThat(handler.getResult(row, 1)).isNull();
|
||||
assertThat(handler.getResult(call, 1)).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void nullInstantBindsAsSqlNullTimestamp() throws Exception {
|
||||
List<Integer> nullTypes = new ArrayList<>();
|
||||
PreparedStatement statement = (PreparedStatement) Proxy.newProxyInstance(getClass().getClassLoader(),
|
||||
new Class<?>[]{PreparedStatement.class}, (proxy, method, args) -> {
|
||||
if (!method.getName().equals("setNull")) {
|
||||
throw new UnsupportedOperationException(method.getName());
|
||||
}
|
||||
nullTypes.add((Integer) args[1]);
|
||||
return null;
|
||||
});
|
||||
snapshotHandler(configuration()).setParameter(statement, 1, null, org.apache.ibatis.type.JdbcType.TIMESTAMP);
|
||||
assertThat(nullTypes).containsExactly(java.sql.Types.TIMESTAMP);
|
||||
}
|
||||
|
||||
private List<LocalDateTime> boundTimestamps(Configuration configuration, String id,
|
||||
Map<String, Object> values) throws Exception {
|
||||
List<LocalDateTime> timestamps = new ArrayList<>();
|
||||
PreparedStatement statement = (PreparedStatement) Proxy.newProxyInstance(getClass().getClassLoader(),
|
||||
new Class<?>[]{PreparedStatement.class}, (proxy, method, args) -> {
|
||||
if (method.getName().equals("setTimestamp")) {
|
||||
Calendar calendar = args.length == 3 ? (Calendar) args[2] : Calendar.getInstance();
|
||||
timestamps.add(LocalDateTime.ofInstant(((Timestamp) args[1]).toInstant(),
|
||||
calendar.getTimeZone().toZoneId()));
|
||||
} else if (!method.getName().startsWith("set")) {
|
||||
throw new UnsupportedOperationException(method.getName());
|
||||
}
|
||||
return null;
|
||||
});
|
||||
MappedStatement mapped = configuration.getMappedStatement(NAMESPACE + id);
|
||||
new DefaultParameterHandler(mapped, values, mapped.getBoundSql(values)).setParameters(statement);
|
||||
return timestamps;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private TypeHandler<Instant> snapshotHandler(Configuration configuration) {
|
||||
return (TypeHandler<Instant>) configuration.getMappedStatement(NAMESPACE + "selectPublishedSnapshot")
|
||||
.getResultMaps().get(0).getResultMappings().stream()
|
||||
.filter(mapping -> "synchronizedAt".equals(mapping.getProperty()))
|
||||
.<TypeHandler<?>>map(mapping -> mapping.getTypeHandler()).findFirst()
|
||||
.orElse(configuration.getTypeHandlerRegistry().getTypeHandler(Instant.class));
|
||||
}
|
||||
|
||||
private Configuration configuration() throws Exception {
|
||||
Configuration configuration = new Configuration();
|
||||
String resource = "mapper/algorithm/ShadowInputMapper.xml";
|
||||
try (InputStream stream = getClass().getClassLoader().getResourceAsStream(resource)) {
|
||||
new XMLMapperBuilder(stream, configuration, resource, configuration.getSqlFragments()).parse();
|
||||
}
|
||||
return configuration;
|
||||
}
|
||||
|
||||
private ShadowSnapshotCoordinator.Admission admission(Instant time, Instant now) {
|
||||
ShadowSnapshotCoordinator coordinator = new ShadowSnapshotCoordinator("01", Duration.ofSeconds(60),
|
||||
Clock.fixed(now, ZoneOffset.UTC));
|
||||
var ticket = coordinator.tryBeginRefresh(now).orElseThrow();
|
||||
assertThat(coordinator.publish(ticket, new PublishedShadowSnapshot(453L, "01", time, "ABC"), now))
|
||||
.isTrue();
|
||||
return coordinator.enter(now);
|
||||
}
|
||||
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
private Timestamp driverTimestamp(Calendar calendar) throws Exception {
|
||||
Class<? extends Enum> jdbcType = (Class<? extends Enum>) Class.forName("com.microsoft.sqlserver.jdbc.JDBCType");
|
||||
Class<? extends Enum> sqlType = (Class<? extends Enum>) Class.forName("com.microsoft.sqlserver.jdbc.SSType");
|
||||
Class<?> connection = Class.forName("com.microsoft.sqlserver.jdbc.SQLServerConnection");
|
||||
// The real driver's Calendar-aware datetime2 decoder, without opening a database connection.
|
||||
Method conversion = Class.forName("com.microsoft.sqlserver.jdbc.DDC").getDeclaredMethod(
|
||||
"convertTemporalToObject", connection, jdbcType, sqlType, Calendar.class,
|
||||
int.class, long.class, int.class);
|
||||
conversion.setAccessible(true);
|
||||
return (Timestamp) conversion.invoke(null, null, Enum.valueOf(jdbcType, "TIMESTAMP"),
|
||||
Enum.valueOf(sqlType, "DATETIME2"), calendar,
|
||||
(int) ChronoUnit.DAYS.between(LocalDate.of(1, 1, 1), SQL_TIME.toLocalDate()),
|
||||
SQL_TIME.toLocalTime().toNanoOfDay(), 7);
|
||||
}
|
||||
}
|
||||
@@ -167,6 +167,30 @@ class ShadowV3OperationsPackageTest {
|
||||
assertThat(export.output()).contains("read-only SQL export failed");
|
||||
}
|
||||
|
||||
@Test
|
||||
void exportUsesLiteralSqlcmd13OutputPathsAndRestoresCallerDirectory() throws Exception {
|
||||
Assumptions.assumeTrue(isWindows(), "CMD behavior is verified on Windows");
|
||||
Path fixture = Files.createDirectories(tempDir.resolve("sqlcmd13 fixture with spaces"));
|
||||
Path bin = writeSqlcmd13Fixture(fixture);
|
||||
Path logDir = Files.createDirectories(fixture.resolve("logs with spaces"));
|
||||
Files.writeString(logDir.resolve("shadow-diagnostics.log"),
|
||||
"2026-09-07T01:31:01Z WARN shadow_failure stage=REFRESH sqlCode=1205\n");
|
||||
Path output = fixture.resolve("export with spaces");
|
||||
CmdResult result = runCmd(root.resolve("scripts/shadow-v3-export.cmd"), bin,
|
||||
root.toString(), "SQL SERVER WITH SPACE", "FT546 WMS", "2026-09-07T01:00:00Z",
|
||||
output.toString(), logDir.toString());
|
||||
assertThat(result.exitCode()).as(result.output()).isZero();
|
||||
assertThat(result.output()).contains("EXPORT_OK");
|
||||
for (String filename : new String[]{"sql-diagnostics.txt", "comparison-detail.txt",
|
||||
"runtime-metrics.txt", "inventory-parity.txt"}) {
|
||||
assertThat(Files.readString(output.resolve(filename))).contains("read-only SQL output fixture");
|
||||
}
|
||||
assertThat(Path.of(Files.readString(output.resolve("sqlcmd-cwd.txt"))))
|
||||
.isEqualTo(output);
|
||||
assertThat(Files.readString(output.resolve("shadow-warnings.txt")))
|
||||
.contains("shadow_failure", "sqlCode=1205");
|
||||
}
|
||||
|
||||
@Test
|
||||
void exportRefusesToOverwriteExistingEvidence() throws Exception {
|
||||
Assumptions.assumeTrue(isWindows(), "CMD behavior is verified on Windows");
|
||||
@@ -197,6 +221,21 @@ class ShadowV3OperationsPackageTest {
|
||||
return bin;
|
||||
}
|
||||
|
||||
private Path writeSqlcmd13Fixture(Path fixture) throws Exception {
|
||||
Path bin = Files.createDirectories(fixture.resolve("stub tools"));
|
||||
Path compiler = Path.of(System.getenv("SystemRoot"), "Microsoft.NET", "Framework64", "v4.0.30319", "csc.exe");
|
||||
if (!Files.isRegularFile(compiler)) {
|
||||
compiler = Path.of(System.getenv("SystemRoot"), "Microsoft.NET", "Framework", "v4.0.30319", "csc.exe");
|
||||
}
|
||||
assertThat(compiler).as("Windows .NET Framework compiler for no-database CMD fixture").isRegularFile();
|
||||
Path executable = bin.resolve("sqlcmd.exe");
|
||||
CmdResult compiled = runProcess(new ProcessBuilder(compiler.toString(), "/nologo",
|
||||
"/out:" + executable, root.resolve("ruoyi-admin/src/test/resources/shadow/Sqlcmd13OutputFixture.cs").toString()), 30);
|
||||
assertThat(compiled.exitCode()).as(compiled.output()).isZero();
|
||||
Files.copy(executable, bin.resolve("jps.exe"));
|
||||
return bin;
|
||||
}
|
||||
|
||||
private void cloneRepository(Path destination) throws Exception {
|
||||
CmdResult result = runProcess(new ProcessBuilder(
|
||||
"git", "clone", "--quiet", root.toString(), destination.toString()), 60);
|
||||
@@ -215,23 +254,38 @@ class ShadowV3OperationsPackageTest {
|
||||
for (String argument : arguments) {
|
||||
command.append(' ').append(quote(argument));
|
||||
}
|
||||
ProcessBuilder builder = new ProcessBuilder("cmd.exe", "/d", "/c", command.toString());
|
||||
Path restored = Files.createTempFile(tempDir, "caller-directory-", ".txt");
|
||||
Path wrapper = Files.createTempFile(tempDir, "caller-", ".cmd");
|
||||
boolean exportScript = script.getFileName().toString().equals("shadow-v3-export.cmd");
|
||||
Path callerDirectory = exportScript ? root : tempDir;
|
||||
Files.writeString(wrapper, "@echo off\r\n"
|
||||
+ (exportScript ? "pushd " + quote(root.toString()) + "\r\n" : "")
|
||||
+ command + "\r\nset \"RESULT=%errorlevel%\"\r\n"
|
||||
+ "cd > " + quote(restored.toString()) + "\r\n"
|
||||
+ (exportScript ? "popd\r\n" : "") + "exit /b %RESULT%\r\n");
|
||||
ProcessBuilder builder = new ProcessBuilder("cmd.exe", "/d", "/c", "call " + quote(wrapper.toString()));
|
||||
builder.directory(tempDir.toFile());
|
||||
String inheritedPath = builder.environment().getOrDefault("PATH", "");
|
||||
builder.environment().put("PATH", stubBin + File.pathSeparator + inheritedPath);
|
||||
builder.environment().put("FORTUCKY_ALGORITHM_WAREHOUSE_ID", "01");
|
||||
builder.redirectErrorStream(true);
|
||||
return runProcess(builder, 15);
|
||||
CmdResult result = runProcess(builder, 15);
|
||||
assertThat(Path.of(Files.readString(restored).trim())).isEqualTo(callerDirectory);
|
||||
return result;
|
||||
}
|
||||
|
||||
private CmdResult runProcess(ProcessBuilder builder, int timeoutSeconds) throws Exception {
|
||||
builder.redirectErrorStream(true);
|
||||
Path capturedOutput = Files.createTempFile(tempDir, "process-output-", ".txt");
|
||||
builder.redirectOutput(capturedOutput.toFile());
|
||||
Process process = builder.start();
|
||||
boolean finished = process.waitFor(timeoutSeconds, TimeUnit.SECONDS);
|
||||
if (!finished) {
|
||||
process.destroyForcibly();
|
||||
process.descendants().forEach(ProcessHandle::destroyForcibly);
|
||||
process.destroyForcibly().waitFor(5, TimeUnit.SECONDS);
|
||||
}
|
||||
assertThat(finished).as("CMD script completed").isTrue();
|
||||
String output = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
|
||||
String output = new String(Files.readAllBytes(capturedOutput), StandardCharsets.UTF_8);
|
||||
assertThat(finished).as("CMD script completed: %s", output).isTrue();
|
||||
return new CmdResult(process.exitValue(), output);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
using System;
|
||||
using System.IO;
|
||||
|
||||
// No SQL execution. Model sqlcmd 13's literal :out paths at the process boundary.
|
||||
class Sqlcmd13OutputFixture
|
||||
{
|
||||
static int Main(string[] args)
|
||||
{
|
||||
if (Path.GetFileNameWithoutExtension(Environment.GetCommandLineArgs()[0]) == "jps")
|
||||
{
|
||||
Console.WriteLine("123 sun.tools.jps.Jps -lv");
|
||||
return 0;
|
||||
}
|
||||
int input = Array.IndexOf(args, "-i");
|
||||
if (input < 0) return 86;
|
||||
foreach (string line in File.ReadAllLines(args[input + 1]))
|
||||
{
|
||||
if (!line.StartsWith(":out ")) continue;
|
||||
string destination = line.Substring(5).Trim().Trim('"');
|
||||
if (destination == "stdout") continue;
|
||||
if (destination.Contains("$("))
|
||||
{
|
||||
Console.Error.WriteLine("sqlcmd13 cannot expand :out destination: " + destination);
|
||||
return 13;
|
||||
}
|
||||
File.WriteAllText(destination, "read-only SQL output fixture\r\n");
|
||||
}
|
||||
File.WriteAllText("sqlcmd-cwd.txt", Environment.CurrentDirectory);
|
||||
Console.WriteLine("EXPORT_SQL_OK");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user