form-center/form-center-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Define authoritative reconciliation acknowledgements

Browse Source 
Local review candidate only. Explicitly selected inherited prerequisites and prior repair patches; unrelated WIP and independent MQ fixes are excluded. No push or deployment authorized.
This commit is contained in:
chenxing
2026-10-23 10:51:16 +08:00
parent c6e061815a
commit 6ff39b0cec
9 changed files with 159 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/yunhai/formcenter/model/FormSyncModels.java
+4
View File
@@ -214,6 +214,10 @@ public final class FormSyncModels {
/** 比中心已记录的最新 attempt 更旧：留档但不推进状态 */
private Boolean staleAttempt;
private String taskState;
/** CURRENT_ACCEPTED / CURRENT_DUPLICATE / STALE_RETRY / CENTER_TERMINAL */
private String decision;
private Integer reportAttempt;
private String resultState;
private String message;
}
src/main/java/com/yunhai/formcenter/service/AnonReplayGuard.java
+10 -3
View File
@@ -45,9 +45,16 @@ public class AnonReplayGuard {
private static final char[] HEX = "0123456789abcdef".toCharArray();
private final String secret;
private final java.time.Clock clock;
@org.springframework.beans.factory.annotation.Autowired
public AnonReplayGuard(@Value("${yh.app-auth.app-secret:}") String secret) {
this(secret, java.time.Clock.systemUTC());
}
AnonReplayGuard(String secret, java.time.Clock clock) {
this.secret = secret == null ? "" : secret.trim();
this.clock = java.util.Objects.requireNonNull(clock);
}
public static String cookieName(String formCode) {
@@ -66,7 +73,7 @@ public class AnonReplayGuard {
*/
public String issue(String formCode) {
requireSecret();
return issuedAt(formCode, System.currentTimeMillis() / 1000L);
return issuedAt(formCode, clock.instant().getEpochSecond());
}
/**
@@ -89,7 +96,7 @@ public class AnonReplayGuard {
if (issuedAt == null) {
return 0L;
}
long remaining = issuedAt + window - System.currentTimeMillis() / 1000L;
long remaining = issuedAt + window - clock.instant().getEpochSecond();
return remaining > 0 ? remaining : 0L;
}
@@ -138,7 +145,7 @@ public class AnonReplayGuard {
}
// 未来时间戳按"刚提交过"处理（窗口不越过配置值）：签名有效说明它是本机签发的，
// 时钟抖动/机器时间偏快都不该把窗口拉长，更不该反过来把过去的票据顶成"刚刚"
return Math.min(issuedAt, System.currentTimeMillis() / 1000L);
return Math.min(issuedAt, clock.instant().getEpochSecond());
}
/** 签发指定制表时刻的票据值（窗口边界用例要用"过去的时间戳"，包内可见即可，不对外暴露） */
src/main/java/com/yunhai/formcenter/service/AnswerReconcileService.java
+26 -24
View File
@@ -736,38 +736,33 @@ public class AnswerReconcileService {
"回报的 attempt=" + attempt + " 超过中心已派发的 " + dispatchedAttempt);
}
Optional<AnswerReconcileResult> repeat = resultRepository.findByTaskIdAndAttempt(task.getId(), attempt);
if (AnswerReconcileTask.isTerminal(task.getState())
|| (repeat.isPresent() && !AnswerReconcileTask.STATE_RUNNING.equals(repeat.get().getState()))) {
ResultAck ack = toResultAck(task);
ack.setAccepted(true);
ack.setDuplicate(true);
ack.setMessage("任务或本轮结果已终结，仅幂等确认，不恢复执行");
return ack;
Integer latest = resultRepository.maxAttempt(task.getId());
boolean stale = attempt < dispatchedAttempt || (latest != null && attempt < latest);
if (AnswerReconcileTask.isTerminal(task.getState())) {
return resultAck(task, attempt, "CENTER_TERMINAL", true, stale, null);
}
String reportState = normalizeReportState(report.getState());
if (stale) {
// 旧轮结论只能归档；重复旧终态也不能被后来回报改写。
boolean duplicate = repeat.isPresent() && !AnswerReconcileTask.STATE_RUNNING.equals(repeat.get().getState());
if (!duplicate) persistResult(task, attempt, report);
return resultAck(task, attempt, "STALE_RETRY", duplicate, true,
duplicate ? repeat.get().getState() : reportState);
}
if (repeat.isPresent() && !AnswerReconcileTask.STATE_RUNNING.equals(repeat.get().getState())) {
return resultAck(task, attempt, "CURRENT_DUPLICATE", true, false, repeat.get().getState());
}
if (AnswerReconcileTask.STATE_COMPLETED.equals(reportState)
&& (!Boolean.TRUE.equals(report.getManifestComplete())
|| !Boolean.TRUE.equals(report.getLocalScanComplete())
|| !Integer.valueOf(1).equals(task.getManifestComplete()))) {
throw new BizException(ErrorCode.PARAM_ERROR, "INCOMPLETE_RESULT：两侧扫描未完整结束，不能回报COMPLETED");
}
Integer latest = resultRepository.maxAttempt(task.getId());
boolean staleAttempt = attempt < dispatchedAttempt || (latest != null && attempt < latest);
if (repeat.isPresent() && AnswerReconcileTask.STATE_RUNNING.equals(reportState) && !staleAttempt) {
advanceByReport(task, report, repeat.get());
return toResultAck(taskRepository.findById(task.getId()).orElse(task));
}
AnswerReconcileResult row = persistResult(task, attempt, report);
ResultAck ack = toResultAck(task);
ack.setAccepted(true);
ack.setDuplicate(false);
ack.setStaleAttempt(staleAttempt);
if (staleAttempt) {
ack.setMessage("旧 attempt=" + attempt + " 的回报只留档，不推进任务状态（最新 attempt=" + latest + "）");
return ack;
}
AnswerReconcileResult row = repeat.isPresent() && AnswerReconcileTask.STATE_RUNNING.equals(reportState)
? repeat.get() : persistResult(task, attempt, report);
advanceByReport(task, report, row);
return toResultAck(taskRepository.findById(task.getId()).orElse(task));
return resultAck(taskRepository.findById(task.getId()).orElse(task), attempt,
"CURRENT_ACCEPTED", false, false, reportState);
}
private AnswerReconcileResult persistResult(AnswerReconcileTask task, int attempt, ResultReport report) {
@@ -866,11 +861,18 @@ public class AnswerReconcileService {
return out;
}
private ResultAck toResultAck(AnswerReconcileTask task) {
private ResultAck resultAck(AnswerReconcileTask task, int reportAttempt, String decision,
boolean duplicate, boolean stale, String resultState) {
ResultAck ack = new ResultAck();
ack.setTaskId(task.getTaskUuid());
ack.setAttempt(task.getAttempt());
ack.setReportAttempt(reportAttempt);
ack.setTaskState(task.getState());
ack.setDecision(decision);
ack.setAccepted(true);
ack.setDuplicate(duplicate);
ack.setStaleAttempt(stale);
ack.setResultState(resultState);
return ack;
}
src/test/java/com/yunhai/formcenter/service/AnonReplayGuardTest.java
+5 -4
View File
@@ -21,7 +21,8 @@ class AnonReplayGuardTest {
private static final String SECRET = "unit-test-secret";
private static final String CODE = "school_survey";
private final AnonReplayGuard guard = new AnonReplayGuard(SECRET);
private final java.time.Clock clock = java.time.Clock.fixed(java.time.Instant.parse("2026-10-23T00:00:00Z"), java.time.ZoneOffset.UTC);
private final AnonReplayGuard guard = new AnonReplayGuard(SECRET, clock);
@Test
@DisplayName("票据名按表单隔离：前缀 fc_anon_ + code（一表单一份，互不干扰）")
@@ -48,14 +49,14 @@ class AnonReplayGuardTest {
// 1 分钟窗口下，30 秒前签发的票据还剩约 30 秒，文案向上取整至少 1 分钟
long remaining = 30L;
assertThat(AnonReplayGuard.remainingMinutes(remaining)).isEqualTo(1);
assertThat(guard.throttleSeconds(CODE, 1, guard.issuedAt(CODE, System.currentTimeMillis() / 1000L - 30)))
assertThat(guard.throttleSeconds(CODE, 1, guard.issuedAt(CODE, clock.instant().getEpochSecond() - 30)))
.isBetween(29L, 31L);
}
@Test
@DisplayName("窗口已过=放行；未来时间戳按\"刚提交过\"处理（窗口不拉长，也不反过来把旧票据顶成刚刚）")
void expiredTicketPassesButFutureOneThrottles() {
long now = System.currentTimeMillis() / 1000L;
long now = clock.instant().getEpochSecond();
assertThat(guard.throttleSeconds(CODE, 10, guard.issuedAt(CODE, now - 601))).isZero();
assertThat(guard.throttleSeconds(CODE, 10, guard.issuedAt(CODE, now - 300))).isBetween(299L, 301L);
assertThat(guard.throttleSeconds(CODE, 10, guard.issuedAt(CODE, now + 120))).isBetween(595L, 600L);
@@ -64,7 +65,7 @@ class AnonReplayGuardTest {
@Test
@DisplayName("伪造无效：换密钥签的值、改过时间戳的值、别的表单的值、形态非法的值，全部按没带 cookie 处理")
void forgedOrMalformedValueIsTreatedAsAbsent() {
long now = System.currentTimeMillis() / 1000L;
long now = clock.instant().getEpochSecond();
String valid = guard.issue(CODE);
// 换密钥（攻击者不知道服务端密钥时能造出的最强形态）
src/test/java/com/yunhai/formcenter/service/FcReconcileTaskTest.java
+55
View File
@@ -603,6 +603,9 @@ class FcReconcileTaskTest {
report(1, AnswerReconcileTask.STATE_FAILED));
assertThat(ack.getStaleAttempt()).isTrue();
assertThat(ack.getDecision()).isEqualTo("STALE_RETRY");
assertThat(ack.getReportAttempt()).isEqualTo(1);
assertThat(ack.getAttempt()).isEqualTo(3);
assertThat(ack.getAccepted()).isTrue();
verify(resultRepository).saveAndFlush(any(AnswerReconcileResult.class));
verify(taskRepository, never()).advanceTerminal(anyLong(), anyString(), any(), any(), any());
@@ -612,6 +615,58 @@ class FcReconcileTaskTest {
assertThat(task.getState()).isEqualTo(AnswerReconcileTask.STATE_RUNNING);
}
@Test
void oldReportCannotOverwriteNewTerminalDecision() {
AnswerReconcileTask task = task(AnswerReconcileTask.STATE_FAILED);
task.setAttempt(3);
when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
.thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
ResultAck ack = service.submitResult(caller(CONSUMER), TASK_ID, report(1, AnswerReconcileTask.STATE_COMPLETED));
assertThat(ack.getDecision()).isEqualTo("CENTER_TERMINAL");
assertThat(ack.getTaskState()).isEqualTo(AnswerReconcileTask.STATE_FAILED);
assertThat(ack.getStaleAttempt()).isTrue();
assertThat(ack.getAttempt()).isEqualTo(3);
assertThat(ack.getReportAttempt()).isEqualTo(1);
verify(resultRepository, never()).saveAndFlush(any());
verify(taskRepository, never()).finishAttempt(anyLong(), anyInt(), anyString(), any(), any(), any());
}
@Test
void duplicateFailureIsNotRewrittenByLaterSuccessClaim() {
AnswerReconcileTask task = task(AnswerReconcileTask.STATE_RUNNING);
task.setAttempt(2);
when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
when(resultRepository.findByTaskIdAndAttempt(1L, 2))
.thenReturn(Optional.of(result(2, AnswerReconcileTask.STATE_FAILED)));
when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
.thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
ResultAck ack = service.submitResult(caller(CONSUMER), TASK_ID, report(2, AnswerReconcileTask.STATE_COMPLETED));
assertThat(ack.getDecision()).isEqualTo("CURRENT_DUPLICATE");
assertThat(ack.getResultState()).isEqualTo(AnswerReconcileTask.STATE_FAILED);
assertThat(ack.getDuplicate()).isTrue();
assertThat(ack.getReportAttempt()).isEqualTo(2);
verify(resultRepository, never()).saveAndFlush(any());
}
@Test
void duplicateOldTerminalResultStillRequiresNewRoundRetry() {
AnswerReconcileTask task = task(AnswerReconcileTask.STATE_RUNNING);
task.setAttempt(3);
when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
when(resultRepository.findByTaskIdAndAttempt(1L, 1))
.thenReturn(Optional.of(result(1, AnswerReconcileTask.STATE_COMPLETED)));
when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
.thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
ResultAck ack = service.submitResult(caller(CONSUMER), TASK_ID, report(1, AnswerReconcileTask.STATE_FAILED));
assertThat(ack.getDecision()).isEqualTo("STALE_RETRY");
assertThat(ack.getStaleAttempt()).isTrue();
assertThat(ack.getResultState()).isEqualTo(AnswerReconcileTask.STATE_COMPLETED);
assertThat(ack.getTaskState()).isEqualTo(AnswerReconcileTask.STATE_RUNNING);
verify(resultRepository, never()).saveAndFlush(any());
verify(taskRepository, never()).renewRunning(anyLong(), anyInt(), any());
}
@Test
@DisplayName("假 result：回报 attempt 超过中心已派发数即拒；跨消费方身份同样拒")
void resultFromUnknownAttemptOrCallerIsRejected() {