From 0d09eb93da24c1b792d7d8a24a2c0c49d760a551 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Tue, 11 Aug 2026 10:39:15 +0800 Subject: [PATCH] feat(wcs): deliver device commands through outbox --- .../device/CommandDeliveryRecorder.java | 111 +++++++ .../device/CommandOutboxDispatcher.java | 58 ++++ .../device/DeviceCommandEnvelope.java | 9 +- .../device/DeviceExecutionService.java | 93 ++++++ .../device/OutboxDispatchResult.java | 35 +++ .../wcs/application/device/SendReceipt.java | 7 +- .../WmsNotificationOutboxDispatcher.java | 77 +++++ .../WmsNotificationOutboxProjector.java | 25 ++ .../domain/execution/CommandOutboxEntry.java | 65 +++- .../domain/execution/OutboxDeliveryState.java | 1 + .../execution/WmsNotificationOutboxEntry.java | 54 +++- .../wcs/port/out/CommandOutboxRepository.java | 4 +- .../out/WmsNotificationOutboxRepository.java | 4 +- .../device/CommandOutboxDispatcherTest.java | 278 ++++++++++++++++++ .../WmsNotificationOutboxDispatcherTest.java | 130 ++++++++ 15 files changed, 941 insertions(+), 10 deletions(-) create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandDeliveryRecorder.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcher.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/OutboxDispatchResult.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcher.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxProjector.java create mode 100644 ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java create mode 100644 ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcherTest.java diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandDeliveryRecorder.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandDeliveryRecorder.java new file mode 100644 index 0000000..c3b66d1 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandDeliveryRecorder.java @@ -0,0 +1,111 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.CommandOutboxEntry; +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.domain.execution.DeviceCommandState; +import com.ruoyi.wcs.port.out.CommandOutboxRepository; +import com.ruoyi.wcs.port.out.DeviceCommandRepository; +import com.ruoyi.wcs.port.out.RecoveryCaseOpener; +import com.ruoyi.wcs.port.out.ResourceReservationRepository; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Instant; +import java.util.Collections; +import java.util.Objects; + +public final class CommandDeliveryRecorder { + private final DeviceCommandRepository commandRepository; + private final CommandOutboxRepository outboxRepository; + private final RecoveryCaseOpener recoveryCaseOpener; + private final ResourceReservationRepository reservationRepository; + + public CommandDeliveryRecorder(DeviceCommandRepository commandRepository, + CommandOutboxRepository outboxRepository, + RecoveryCaseOpener recoveryCaseOpener, + ResourceReservationRepository reservationRepository) { + this.commandRepository = Objects.requireNonNull(commandRepository, + "commandRepository"); + this.outboxRepository = Objects.requireNonNull(outboxRepository, + "outboxRepository"); + this.recoveryCaseOpener = Objects.requireNonNull(recoveryCaseOpener, + "recoveryCaseOpener"); + this.reservationRepository = Objects.requireNonNull(reservationRepository, + "reservationRepository"); + } + + @Transactional + public OutboxDispatchResult.Outcome record(CommandOutboxEntry entry, + SendReceipt receipt, + Instant now) { + Objects.requireNonNull(entry, "entry"); + Objects.requireNonNull(receipt, "receipt"); + Objects.requireNonNull(now, "now"); + DeviceCommand command = commandRepository.findByCommandId( + entry.getCommandId()); + if (command == null || command.getState() != DeviceCommandState.PENDING) { + entry.markDead("COMMAND_NOT_PENDING", now); + outboxRepository.save(entry); + return OutboxDispatchResult.Outcome.DEAD; + } + + switch (receipt.getStatus()) { + case NOT_SENT: + entry.reschedule(detail(receipt, "NOT_SENT"), + now.plusSeconds(backoffSeconds(entry.getAttemptCount()))); + outboxRepository.save(entry); + return OutboxDispatchResult.Outcome.RETRY_SCHEDULED; + case ACCEPTED: + command.markSent(receipt.getReceiptId(), now); + entry.markDelivered(now); + commandRepository.save(command); + outboxRepository.save(entry); + return OutboxDispatchResult.Outcome.SENT; + case ACCEPTED_UNKNOWN: + String reason = detail(receipt, "WRITE_RESULT_AMBIGUOUS"); + command.markSent(receipt.getReceiptId(), now); + command.markUnknown(reason, now); + entry.markDelivered(now); + reservationRepository.protectForUnknown(command.getCommandKey(), + receipt.getReceiptId()); + recoveryCaseOpener.open(command.getDeviceId(), + command.getCommandKey(), reason, + Collections.singletonList(receipt.getReceiptId())); + commandRepository.save(command); + outboxRepository.save(entry); + return OutboxDispatchResult.Outcome.ISOLATED_UNKNOWN; + case REJECTED: + command.markSent(receipt.getReceiptId(), now); + command.reject(detail(receipt, "DEVICE_REJECTED"), now); + entry.markDelivered(now); + commandRepository.save(command); + outboxRepository.save(entry); + return OutboxDispatchResult.Outcome.REJECTED; + default: + throw new IllegalStateException("unsupported send status"); + } + } + + DeviceCommand findPendingCommand(String commandId) { + DeviceCommand command = commandRepository.findByCommandId(commandId); + return command != null && command.getState() == DeviceCommandState.PENDING + ? command : null; + } + + @Transactional + public OutboxDispatchResult.Outcome discardInvalid( + CommandOutboxEntry entry, Instant now) { + entry.markDead("COMMAND_NOT_PENDING", now); + outboxRepository.save(entry); + return OutboxDispatchResult.Outcome.DEAD; + } + + private static long backoffSeconds(int attemptCount) { + int exponent = Math.min(Math.max(attemptCount, 1), 8); + return Math.min(300L, 1L << exponent); + } + + private static String detail(SendReceipt receipt, String fallback) { + String detail = receipt.getDetail(); + return detail == null || detail.trim().isEmpty() ? fallback : detail.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcher.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcher.java new file mode 100644 index 0000000..88b4c8f --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcher.java @@ -0,0 +1,58 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.CommandOutboxEntry; +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.port.out.CommandOutboxRepository; +import com.ruoyi.wcs.port.out.DeviceCommandGateway; + +import java.time.Instant; +import java.util.List; +import java.util.Objects; + +public final class CommandOutboxDispatcher { + private final CommandOutboxRepository outboxRepository; + private final DeviceCommandGateway commandGateway; + private final CommandDeliveryRecorder recorder; + + public CommandOutboxDispatcher(CommandOutboxRepository outboxRepository, + DeviceCommandGateway commandGateway, + CommandDeliveryRecorder recorder) { + this.outboxRepository = Objects.requireNonNull(outboxRepository, + "outboxRepository"); + this.commandGateway = Objects.requireNonNull(commandGateway, + "commandGateway"); + this.recorder = Objects.requireNonNull(recorder, "recorder"); + } + + public OutboxDispatchResult dispatchDue(int limit, Instant now) { + if (limit <= 0) { + throw new IllegalArgumentException("limit must be positive"); + } + Objects.requireNonNull(now, "now"); + List entries = outboxRepository.claimDue(limit, now); + OutboxDispatchResult result = new OutboxDispatchResult(); + for (CommandOutboxEntry entry : entries) { + DeviceCommand command = recorder.findPendingCommand(entry.getCommandId()); + if (command == null) { + result.record(recorder.discardInvalid(entry, now)); + continue; + } + SendReceipt receipt; + try { + receipt = commandGateway.send(DeviceCommandEnvelope.from(command, + entry.getAttemptCount(), entry.getTopologyVersion(), + entry.getFencingToken())); + if (receipt == null) { + throw new IllegalStateException("gateway returned null receipt"); + } + } catch (RuntimeException ambiguous) { + receipt = SendReceipt.acceptedUnknown( + "AMBIGUOUS:" + entry.getCommandId() + ":" + + entry.getAttemptCount(), + "GATEWAY_EXCEPTION_AFTER_SEND_ATTEMPT"); + } + result.record(recorder.record(entry, receipt, now)); + } + return result; + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java index d642f9a..a37f88f 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceCommandEnvelope.java @@ -33,9 +33,16 @@ public final class DeviceCommandEnvelope { public static DeviceCommandEnvelope from(DeviceCommand command, long topologyVersion, long fencingToken) { + return from(command, command.getAttempt(), topologyVersion, fencingToken); + } + + public static DeviceCommandEnvelope from(DeviceCommand command, + int deliveryAttempt, + long topologyVersion, + long fencingToken) { return new DeviceCommandEnvelope(command.getCommandId(), command.getCommandKey(), command.getDeviceId(), command.getType(), - command.getPayload(), command.getAttempt(), topologyVersion, + command.getPayload(), deliveryAttempt, topologyVersion, fencingToken); } diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java new file mode 100644 index 0000000..5561bd5 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java @@ -0,0 +1,93 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.application.dispatch.GrantedDispatch; +import com.ruoyi.wcs.domain.execution.CommandOutboxEntry; +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.domain.execution.DeviceCommandType; +import com.ruoyi.wcs.domain.shared.DomainRuleViolation; +import com.ruoyi.wcs.domain.shared.IdGenerator; +import com.ruoyi.wcs.port.out.CommandOutboxRepository; +import com.ruoyi.wcs.port.out.DeviceCommandRepository; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Instant; +import java.util.Objects; +import java.util.function.Supplier; + +public final class DeviceExecutionService { + private final DeviceCommandRepository commandRepository; + private final CommandOutboxRepository outboxRepository; + private final IdGenerator idGenerator; + private final Supplier clock; + + public DeviceExecutionService(DeviceCommandRepository commandRepository, + CommandOutboxRepository outboxRepository, + IdGenerator idGenerator, + Supplier clock) { + this.commandRepository = Objects.requireNonNull(commandRepository, + "commandRepository"); + this.outboxRepository = Objects.requireNonNull(outboxRepository, + "outboxRepository"); + this.idGenerator = Objects.requireNonNull(idGenerator, "idGenerator"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + @Transactional + public DeviceCommand createCommand(GrantedDispatch grant) { + Objects.requireNonNull(grant, "grant"); + String commandKey = "DISPATCH:" + requireText(grant.getOrderId(), "orderId") + + ":" + requireText(grant.getStepId(), "stepId"); + String payload = encode(grant); + DeviceCommand existing = commandRepository.findByCommandKey(commandKey); + if (existing != null) { + if (!existing.getDeviceId().equals(grant.getVehicleId()) + || !existing.getPayload().equals(payload)) { + throw new DomainRuleViolation( + "command key reused with a different dispatch grant"); + } + return existing; + } + + Instant now = Objects.requireNonNull(clock.get(), "clock returned null"); + DeviceCommand command = DeviceCommand.pending( + requireText(idGenerator.nextId(), "generated commandId"), + commandKey, grant.getVehicleId(), DeviceCommandType.MOVE_ROUTE, + payload, 1, now); + CommandOutboxEntry entry = CommandOutboxEntry.pending( + command.getCommandId(), command.getCommandKey(), payload, now, + grant.getTopologyVersion(), grant.getFencingToken()); + commandRepository.save(command); + outboxRepository.saveIfAbsent(entry); + return command; + } + + private static String encode(GrantedDispatch grant) { + StringBuilder route = new StringBuilder(); + for (String node : grant.getProtectedRoute().getNodeIds()) { + if (route.length() > 0) { + route.append('>'); + } + route.append(requireSafe(node, "route node")); + } + return "orderId=" + requireSafe(grant.getOrderId(), "orderId") + + "|stepId=" + requireSafe(grant.getStepId(), "stepId") + + "|vehicleId=" + requireSafe(grant.getVehicleId(), "vehicleId") + + "|route=" + route; + } + + private static String requireSafe(String value, String field) { + String text = requireText(value, field); + if (text.indexOf('|') >= 0 || text.indexOf('>') >= 0 + || text.indexOf('=') >= 0) { + throw new IllegalArgumentException(field + " contains reserved delimiter"); + } + return text; + } + + private static String requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(field + " must not be blank"); + } + return value.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/OutboxDispatchResult.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/OutboxDispatchResult.java new file mode 100644 index 0000000..2ef12f7 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/OutboxDispatchResult.java @@ -0,0 +1,35 @@ +package com.ruoyi.wcs.application.device; + +import java.util.EnumMap; +import java.util.Map; + +public final class OutboxDispatchResult { + public enum Outcome { + SENT, + RETRY_SCHEDULED, + ISOLATED_UNKNOWN, + REJECTED, + DEAD + } + + private final Map counts = + new EnumMap(Outcome.class); + + void record(Outcome outcome) { + Integer count = counts.get(outcome); + counts.put(outcome, count == null ? 1 : count + 1); + } + + public int count(Outcome outcome) { + Integer count = counts.get(outcome); + return count == null ? 0 : count; + } + + public int getTotal() { + int total = 0; + for (Integer count : counts.values()) { + total += count; + } + return total; + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java index 925a662..1ae5b97 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SendReceipt.java @@ -16,7 +16,8 @@ public final class SendReceipt { this.status = java.util.Objects.requireNonNull(status, "status"); this.receiptId = receiptId; this.detail = detail; - if ((status == Status.ACCEPTED || status == Status.ACCEPTED_UNKNOWN) + if ((status == Status.ACCEPTED || status == Status.ACCEPTED_UNKNOWN + || status == Status.REJECTED) && isBlank(receiptId)) { throw new IllegalArgumentException( "accepted send receipt must have a receiptId"); @@ -35,8 +36,8 @@ public final class SendReceipt { return new SendReceipt(Status.NOT_SENT, null, detail); } - public static SendReceipt rejected(String detail) { - return new SendReceipt(Status.REJECTED, null, detail); + public static SendReceipt rejected(String receiptId, String detail) { + return new SendReceipt(Status.REJECTED, receiptId, detail); } public Status getStatus() { return status; } diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcher.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcher.java new file mode 100644 index 0000000..e94db15 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcher.java @@ -0,0 +1,77 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.WmsNotificationOutboxEntry; +import com.ruoyi.wcs.port.out.WmsNotificationGateway; +import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository; +import com.ruoyi.wcs.port.out.WmsNotificationReceipt; + +import java.time.Instant; +import java.util.List; +import java.util.Objects; + +public final class WmsNotificationOutboxDispatcher { + private final WmsNotificationOutboxRepository repository; + private final WmsNotificationGateway gateway; + + public WmsNotificationOutboxDispatcher( + WmsNotificationOutboxRepository repository, + WmsNotificationGateway gateway) { + this.repository = Objects.requireNonNull(repository, "repository"); + this.gateway = Objects.requireNonNull(gateway, "gateway"); + } + + public OutboxDispatchResult dispatchDue(int limit, Instant now) { + if (limit <= 0) { + throw new IllegalArgumentException("limit must be positive"); + } + Objects.requireNonNull(now, "now"); + List entries = repository.claimDue(limit, now); + OutboxDispatchResult result = new OutboxDispatchResult(); + for (WmsNotificationOutboxEntry entry : entries) { + WmsNotificationReceipt receipt; + try { + receipt = gateway.notify(entry.getNotification()); + if (receipt == null) { + throw new IllegalStateException("WMS gateway returned null receipt"); + } + } catch (RuntimeException transportFailure) { + receipt = new WmsNotificationReceipt( + WmsNotificationReceipt.Status.RETRYABLE_FAILURE, + "WMS_TRANSPORT_EXCEPTION"); + } + switch (receipt.getStatus()) { + case DELIVERED: + entry.markDelivered(nonBlank(receipt.getDetail(), + "DELIVERED:" + entry.getEventId()), now); + repository.save(entry); + result.record(OutboxDispatchResult.Outcome.SENT); + break; + case RETRYABLE_FAILURE: + entry.reschedule(nonBlank(receipt.getDetail(), + "WMS_RETRYABLE_FAILURE"), + now.plusSeconds(backoffSeconds(entry.getAttemptCount()))); + repository.save(entry); + result.record(OutboxDispatchResult.Outcome.RETRY_SCHEDULED); + break; + case PERMANENT_FAILURE: + entry.markDead(nonBlank(receipt.getDetail(), + "WMS_PERMANENT_FAILURE"), now); + repository.save(entry); + result.record(OutboxDispatchResult.Outcome.DEAD); + break; + default: + throw new IllegalStateException("unsupported WMS receipt"); + } + } + return result; + } + + private static long backoffSeconds(int attemptCount) { + int exponent = Math.min(Math.max(attemptCount, 1), 8); + return Math.min(300L, 1L << exponent); + } + + private static String nonBlank(String value, String fallback) { + return value == null || value.trim().isEmpty() ? fallback : value.trim(); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxProjector.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxProjector.java new file mode 100644 index 0000000..1336403 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxProjector.java @@ -0,0 +1,25 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.WmsNotificationOutboxEntry; +import com.ruoyi.wcs.port.out.WmsNotification; +import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository; + +import java.time.Instant; +import java.util.Objects; +import org.springframework.transaction.annotation.Transactional; + +public final class WmsNotificationOutboxProjector { + private final WmsNotificationOutboxRepository repository; + + public WmsNotificationOutboxProjector( + WmsNotificationOutboxRepository repository) { + this.repository = Objects.requireNonNull(repository, "repository"); + } + + @Transactional + public void project(WmsNotification notification, Instant nextAttemptAt) { + Objects.requireNonNull(notification, "notification"); + repository.saveIfAbsent(WmsNotificationOutboxEntry.pending( + notification.getEventId(), notification, nextAttemptAt)); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommandOutboxEntry.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommandOutboxEntry.java index 37310cc..ee407fa 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommandOutboxEntry.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/CommandOutboxEntry.java @@ -7,23 +7,73 @@ public final class CommandOutboxEntry { private final String commandId; private final String commandKey; private final String payload; + private final long topologyVersion; + private final long fencingToken; private OutboxDeliveryState state; private int attemptCount; private Instant nextAttemptAt; private String lastError; + private Instant deliveredAt; private CommandOutboxEntry(String commandId, String commandKey, - String payload, Instant nextAttemptAt) { + String payload, Instant nextAttemptAt, + long topologyVersion, long fencingToken) { this.commandId = requireText(commandId, "commandId"); this.commandKey = requireText(commandKey, "commandKey"); this.payload = requireText(payload, "payload"); + if (topologyVersion <= 0L || fencingToken <= 0L) { + throw new IllegalArgumentException("outbox versions are invalid"); + } + this.topologyVersion = topologyVersion; + this.fencingToken = fencingToken; this.nextAttemptAt = Objects.requireNonNull(nextAttemptAt, "nextAttemptAt"); this.state = OutboxDeliveryState.PENDING; } public static CommandOutboxEntry pending(String commandId, String commandKey, String payload, Instant nextAttemptAt) { - return new CommandOutboxEntry(commandId, commandKey, payload, nextAttemptAt); + return pending(commandId, commandKey, payload, nextAttemptAt, 1L, 1L); + } + + public static CommandOutboxEntry pending(String commandId, String commandKey, + String payload, Instant nextAttemptAt, + long topologyVersion, + long fencingToken) { + return new CommandOutboxEntry(commandId, commandKey, payload, + nextAttemptAt, topologyVersion, fencingToken); + } + + public boolean claim(Instant now) { + Instant claimTime = Objects.requireNonNull(now, "now"); + if (state != OutboxDeliveryState.PENDING + || nextAttemptAt.isAfter(claimTime)) { + return false; + } + state = OutboxDeliveryState.IN_FLIGHT; + attemptCount++; + return true; + } + + public void markDelivered(Instant at) { + requireInFlight(); + deliveredAt = Objects.requireNonNull(at, "at"); + lastError = null; + state = OutboxDeliveryState.DELIVERED; + } + + public void reschedule(String error, Instant nextAttemptAt) { + requireInFlight(); + lastError = requireText(error, "error"); + this.nextAttemptAt = Objects.requireNonNull(nextAttemptAt, + "nextAttemptAt"); + state = OutboxDeliveryState.PENDING; + } + + public void markDead(String error, Instant at) { + requireInFlight(); + lastError = requireText(error, "error"); + deliveredAt = Objects.requireNonNull(at, "at"); + state = OutboxDeliveryState.DEAD; } public String getCommandId() { @@ -54,6 +104,17 @@ public final class CommandOutboxEntry { return lastError; } + public long getTopologyVersion() { return topologyVersion; } + public long getFencingToken() { return fencingToken; } + public Instant getDeliveredAt() { return deliveredAt; } + + private void requireInFlight() { + if (state != OutboxDeliveryState.IN_FLIGHT) { + throw new com.ruoyi.wcs.domain.shared.DomainRuleViolation( + "outbox entry must be IN_FLIGHT"); + } + } + private static String requireText(String value, String field) { if (value == null || value.trim().isEmpty()) { throw new IllegalArgumentException(field + " must not be blank"); diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/OutboxDeliveryState.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/OutboxDeliveryState.java index 493ce39..db13f22 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/OutboxDeliveryState.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/OutboxDeliveryState.java @@ -2,6 +2,7 @@ package com.ruoyi.wcs.domain.execution; public enum OutboxDeliveryState { PENDING, + IN_FLIGHT, DELIVERED, DEAD } diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/WmsNotificationOutboxEntry.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/WmsNotificationOutboxEntry.java index 3923f40..da93f1d 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/WmsNotificationOutboxEntry.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/execution/WmsNotificationOutboxEntry.java @@ -8,8 +8,12 @@ import java.util.Objects; public final class WmsNotificationOutboxEntry { private final String eventId; private final WmsNotification notification; - private final Instant nextAttemptAt; - private final OutboxDeliveryState state; + private Instant nextAttemptAt; + private OutboxDeliveryState state; + private int attemptCount; + private String lastError; + private String externalReceipt; + private Instant deliveredAt; private WmsNotificationOutboxEntry(String eventId, WmsNotification notification, @@ -29,6 +33,40 @@ public final class WmsNotificationOutboxEntry { return new WmsNotificationOutboxEntry(eventId, notification, nextAttemptAt); } + public boolean claim(Instant now) { + Instant claimTime = Objects.requireNonNull(now, "now"); + if (state != OutboxDeliveryState.PENDING + || nextAttemptAt.isAfter(claimTime)) { + return false; + } + state = OutboxDeliveryState.IN_FLIGHT; + attemptCount++; + return true; + } + + public void markDelivered(String receipt, Instant at) { + requireInFlight(); + externalReceipt = requireText(receipt, "externalReceipt"); + deliveredAt = Objects.requireNonNull(at, "at"); + lastError = null; + state = OutboxDeliveryState.DELIVERED; + } + + public void reschedule(String error, Instant nextAttemptAt) { + requireInFlight(); + lastError = requireText(error, "error"); + this.nextAttemptAt = Objects.requireNonNull(nextAttemptAt, + "nextAttemptAt"); + state = OutboxDeliveryState.PENDING; + } + + public void markDead(String error, Instant at) { + requireInFlight(); + lastError = requireText(error, "error"); + deliveredAt = Objects.requireNonNull(at, "at"); + state = OutboxDeliveryState.DEAD; + } + public String getEventId() { return eventId; } @@ -45,6 +83,18 @@ public final class WmsNotificationOutboxEntry { return state; } + public int getAttemptCount() { return attemptCount; } + public String getLastError() { return lastError; } + public String getExternalReceipt() { return externalReceipt; } + public Instant getDeliveredAt() { return deliveredAt; } + + private void requireInFlight() { + if (state != OutboxDeliveryState.IN_FLIGHT) { + throw new com.ruoyi.wcs.domain.shared.DomainRuleViolation( + "WMS outbox entry must be IN_FLIGHT"); + } + } + private static String requireText(String value, String field) { if (value == null || value.trim().isEmpty()) { throw new IllegalArgumentException(field + " must not be blank"); diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/CommandOutboxRepository.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/CommandOutboxRepository.java index d6d521e..b27e700 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/CommandOutboxRepository.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/CommandOutboxRepository.java @@ -8,5 +8,7 @@ import java.util.List; public interface CommandOutboxRepository { void saveIfAbsent(CommandOutboxEntry entry); - List findDue(int limit, Instant now); + void save(CommandOutboxEntry entry); + + List claimDue(int limit, Instant now); } diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/WmsNotificationOutboxRepository.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/WmsNotificationOutboxRepository.java index d277326..e3841b1 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/WmsNotificationOutboxRepository.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/port/out/WmsNotificationOutboxRepository.java @@ -8,5 +8,7 @@ import java.util.List; public interface WmsNotificationOutboxRepository { void saveIfAbsent(WmsNotificationOutboxEntry entry); - List findDue(int limit, Instant now); + void save(WmsNotificationOutboxEntry entry); + + List claimDue(int limit, Instant now); } diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java new file mode 100644 index 0000000..75cd19a --- /dev/null +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java @@ -0,0 +1,278 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.application.dispatch.GrantedDispatch; +import com.ruoyi.wcs.domain.execution.CommandOutboxEntry; +import com.ruoyi.wcs.domain.execution.DeviceCommand; +import com.ruoyi.wcs.domain.execution.DeviceCommandState; +import com.ruoyi.wcs.domain.execution.OutboxDeliveryState; +import com.ruoyi.wcs.domain.shared.IdGenerator; +import com.ruoyi.wcs.domain.traffic.RoutePlan; +import com.ruoyi.wcs.port.out.CommandOutboxRepository; +import com.ruoyi.wcs.port.out.DeviceCommandGateway; +import com.ruoyi.wcs.port.out.DeviceCommandRepository; +import com.ruoyi.wcs.port.out.RecoveryCaseOpener; +import com.ruoyi.wcs.port.out.ResourceReservationRepository; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CommandOutboxDispatcherTest { + private FakeCommandRepository commands; + private FakeCommandOutboxRepository outbox; + private FakeGateway gateway; + private FakeIsolation isolation; + private DeviceExecutionService service; + private CommandOutboxDispatcher dispatcher; + + @BeforeEach + void setUp() { + commands = new FakeCommandRepository(); + outbox = new FakeCommandOutboxRepository(); + gateway = new FakeGateway(); + isolation = new FakeIsolation(); + IdGenerator ids = new IdGenerator() { + @Override + public String nextId() { + return "CMD-1"; + } + }; + service = new DeviceExecutionService(commands, outbox, ids, + new java.util.function.Supplier() { + @Override + public Instant get() { + return at(0); + } + }); + CommandDeliveryRecorder recorder = new CommandDeliveryRecorder( + commands, outbox, isolation, isolation); + dispatcher = new CommandOutboxDispatcher(outbox, gateway, recorder); + } + + @Test + void persistsCommandAndOutboxBeforeCallingTheGateway() { + DeviceCommand created = service.createCommand(grantedDispatch()); + + assertNotNull(commands.findByCommandId(created.getCommandId())); + assertNotNull(outbox.byCommandId(created.getCommandId())); + assertEquals(0, gateway.sendCount); + } + + @Test + void acceptedSendIsNotMarkedCompleted() { + service.createCommand(grantedDispatch()); + gateway.receipt = SendReceipt.accepted("PLC-WRITE-1"); + + dispatcher.dispatchDue(10, at(0)); + + assertEquals(DeviceCommandState.SENT, + commands.findByCommandId("CMD-1").getState()); + assertEquals(OutboxDeliveryState.DELIVERED, + outbox.byCommandId("CMD-1").getState()); + } + + @Test + void acceptedUnknownIsIsolatedAndNeverAutomaticallyRetried() { + service.createCommand(grantedDispatch()); + gateway.receipt = SendReceipt.acceptedUnknown( + "PLC-WRITE-UNKNOWN-1", "WRITE_RESULT_AMBIGUOUS"); + + dispatcher.dispatchDue(10, at(0)); + dispatcher.dispatchDue(10, at(100)); + + assertEquals(DeviceCommandState.UNKNOWN, + commands.findByCommandId("CMD-1").getState()); + assertEquals(1, gateway.sendCount); + assertEquals(Collections.singletonList("WRITE_RESULT_AMBIGUOUS"), + isolation.reasons); + assertEquals(1, isolation.protectCount); + } + + @Test + void onlyNotSentOutcomeIsAutomaticallyRetried() { + service.createCommand(grantedDispatch()); + gateway.receipt = SendReceipt.notSent("SOCKET_NOT_OPEN"); + + dispatcher.dispatchDue(10, at(0)); + dispatcher.dispatchDue(10, at(1)); + + assertEquals(DeviceCommandState.PENDING, + commands.findByCommandId("CMD-1").getState()); + assertEquals(1, gateway.sendCount); + assertEquals(OutboxDeliveryState.PENDING, + outbox.byCommandId("CMD-1").getState()); + assertTrue(outbox.byCommandId("CMD-1").getNextAttemptAt() + .isAfter(at(1))); + } + + @Test + void retryKeepsCommandKeyButIncrementsTheDeliveryAttempt() { + service.createCommand(grantedDispatch()); + gateway.receipts.add(SendReceipt.notSent("SOCKET_NOT_OPEN")); + gateway.receipts.add(SendReceipt.accepted("PLC-WRITE-2")); + + dispatcher.dispatchDue(10, at(0)); + dispatcher.dispatchDue(10, at(2)); + + assertEquals(Arrays.asList(1, 2), gateway.deliveryAttempts); + assertEquals(DeviceCommandState.SENT, + commands.findByCommandId("CMD-1").getState()); + assertEquals("DISPATCH:ORDER-1:STEP-1", + commands.findByCommandId("CMD-1").getCommandKey()); + } + + @Test + void unexpectedGatewayExceptionIsAmbiguousAndNotRetried() { + service.createCommand(grantedDispatch()); + gateway.failure = new IllegalStateException("connection reset after write"); + + dispatcher.dispatchDue(10, at(0)); + dispatcher.dispatchDue(10, at(100)); + + assertEquals(DeviceCommandState.UNKNOWN, + commands.findByCommandId("CMD-1").getState()); + assertEquals(1, gateway.sendCount); + assertEquals(OutboxDeliveryState.DELIVERED, + outbox.byCommandId("CMD-1").getState()); + } + + @Test + void staleOutboxCannotSendAfterTheCommandAlreadyLeftPending() { + DeviceCommand command = service.createCommand(grantedDispatch()); + command.markSent("ALREADY-SENT", at(0)); + commands.save(command); + + dispatcher.dispatchDue(10, at(0)); + + assertEquals(0, gateway.sendCount); + assertEquals(OutboxDeliveryState.DEAD, + outbox.byCommandId("CMD-1").getState()); + } + + private static GrantedDispatch grantedDispatch() { + RoutePlan route = new RoutePlan(Arrays.asList("A", "B"), + Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B"), + 1, 0, 7L); + return new GrantedDispatch("ORDER-1", "STEP-1", "VEH-1", route, + Collections.singletonList("RES-1"), 7L, 11L); + } + + private static Instant at(long seconds) { + return Instant.EPOCH.plusSeconds(seconds); + } + + private static final class FakeCommandRepository + implements DeviceCommandRepository { + private final Map values = + new LinkedHashMap(); + + @Override + public DeviceCommand findByCommandKey(String commandKey) { + for (DeviceCommand command : values.values()) { + if (command.getCommandKey().equals(commandKey)) { + return command; + } + } + return null; + } + + @Override + public DeviceCommand findByCommandId(String commandId) { + return values.get(commandId); + } + + @Override + public void save(DeviceCommand command) { + values.put(command.getCommandId(), command); + } + } + + private static final class FakeCommandOutboxRepository + implements CommandOutboxRepository { + private final Map values = + new LinkedHashMap(); + + @Override + public void saveIfAbsent(CommandOutboxEntry entry) { + if (!values.containsKey(entry.getCommandId())) { + values.put(entry.getCommandId(), entry); + } + } + + @Override + public void save(CommandOutboxEntry entry) { + values.put(entry.getCommandId(), entry); + } + + @Override + public List claimDue(int limit, Instant now) { + List result = + new ArrayList(); + for (CommandOutboxEntry entry : values.values()) { + if (result.size() >= limit) { + break; + } + if (entry.claim(now)) { + result.add(entry); + } + } + return result; + } + + CommandOutboxEntry byCommandId(String commandId) { + return values.get(commandId); + } + } + + private static final class FakeGateway implements DeviceCommandGateway { + private SendReceipt receipt = SendReceipt.accepted("PLC-WRITE-1"); + private RuntimeException failure; + private int sendCount; + private final List receipts = new ArrayList(); + private final List deliveryAttempts = new ArrayList(); + + @Override + public SendReceipt send(DeviceCommandEnvelope envelope) { + sendCount++; + deliveryAttempts.add(envelope.getAttempt()); + if (failure != null) { + throw failure; + } + return receipts.isEmpty() ? receipt : receipts.remove(0); + } + } + + private static final class FakeIsolation implements RecoveryCaseOpener, + ResourceReservationRepository { + private final List reasons = new ArrayList(); + private int protectCount; + + @Override + public void open(String deviceId, String commandKey, String reason, + List evidenceIds) { + reasons.add(reason); + } + + @Override + public void releasePassedResources(String commandKey, + List resourceKeys, + String releaseEvidenceId, + long fencingToken) { + } + + @Override + public void protectForUnknown(String commandKey, String evidenceId) { + protectCount++; + } + } +} diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcherTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcherTest.java new file mode 100644 index 0000000..86f9ef8 --- /dev/null +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/WmsNotificationOutboxDispatcherTest.java @@ -0,0 +1,130 @@ +package com.ruoyi.wcs.application.device; + +import com.ruoyi.wcs.domain.execution.OutboxDeliveryState; +import com.ruoyi.wcs.domain.execution.WmsNotificationOutboxEntry; +import com.ruoyi.wcs.port.out.WmsNotification; +import com.ruoyi.wcs.port.out.WmsNotificationGateway; +import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository; +import com.ruoyi.wcs.port.out.WmsNotificationReceipt; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class WmsNotificationOutboxDispatcherTest { + + @Test + void duplicateBusinessEventCreatesOneWmsOutboxEntry() { + FakeWmsOutbox repository = new FakeWmsOutbox(); + WmsNotificationOutboxProjector projector = + new WmsNotificationOutboxProjector(repository); + WmsNotification notification = notification("EVENT-1"); + + projector.project(notification, at(0)); + projector.project(notification, at(0)); + + assertEquals(1, repository.values.size()); + } + + @Test + void transportFailureRetriesWithoutChangingTheBusinessEventKey() { + FakeWmsOutbox repository = new FakeWmsOutbox(); + repository.saveIfAbsent(WmsNotificationOutboxEntry.pending( + "EVENT-1", notification("EVENT-1"), at(0))); + FakeWmsGateway gateway = new FakeWmsGateway(); + gateway.receipt = new WmsNotificationReceipt( + WmsNotificationReceipt.Status.RETRYABLE_FAILURE, "HTTP_503"); + WmsNotificationOutboxDispatcher dispatcher = + new WmsNotificationOutboxDispatcher(repository, gateway); + + dispatcher.dispatchDue(10, at(0)); + dispatcher.dispatchDue(10, at(1)); + + WmsNotificationOutboxEntry entry = repository.values.get("EVENT-1"); + assertEquals("EVENT-1", entry.getEventId()); + assertEquals(1, gateway.calls); + assertEquals(OutboxDeliveryState.PENDING, entry.getState()); + } + + @Test + void deliveredCallbackIsNeverSentAgain() { + FakeWmsOutbox repository = new FakeWmsOutbox(); + repository.saveIfAbsent(WmsNotificationOutboxEntry.pending( + "EVENT-1", notification("EVENT-1"), at(0))); + FakeWmsGateway gateway = new FakeWmsGateway(); + gateway.receipt = new WmsNotificationReceipt( + WmsNotificationReceipt.Status.DELIVERED, "WMS-RECEIPT-1"); + WmsNotificationOutboxDispatcher dispatcher = + new WmsNotificationOutboxDispatcher(repository, gateway); + + dispatcher.dispatchDue(10, at(0)); + dispatcher.dispatchDue(10, at(100)); + + assertEquals(1, gateway.calls); + assertEquals(OutboxDeliveryState.DELIVERED, + repository.values.get("EVENT-1").getState()); + assertEquals("WMS-RECEIPT-1", + repository.values.get("EVENT-1").getExternalReceipt()); + } + + private static WmsNotification notification(String eventId) { + return new WmsNotification(eventId, "REQ-1", "ORDER-1", at(0), + "OrderCompleted", "OK", null, + Collections.emptyMap()); + } + + private static Instant at(long seconds) { + return Instant.EPOCH.plusSeconds(seconds); + } + + private static final class FakeWmsOutbox + implements WmsNotificationOutboxRepository { + private final Map values = + new LinkedHashMap(); + + @Override + public void saveIfAbsent(WmsNotificationOutboxEntry entry) { + if (!values.containsKey(entry.getEventId())) { + values.put(entry.getEventId(), entry); + } + } + + @Override + public void save(WmsNotificationOutboxEntry entry) { + values.put(entry.getEventId(), entry); + } + + @Override + public List claimDue(int limit, Instant now) { + List result = + new ArrayList(); + for (WmsNotificationOutboxEntry entry : values.values()) { + if (result.size() >= limit) { + break; + } + if (entry.claim(now)) { + result.add(entry); + } + } + return result; + } + } + + private static final class FakeWmsGateway + implements WmsNotificationGateway { + private WmsNotificationReceipt receipt; + private int calls; + + @Override + public WmsNotificationReceipt notify(WmsNotification notification) { + calls++; + return receipt; + } + } +}