MongoDB cluster with SpringBoot WebFlux duplicates records

Question

I use mongodb cluster with 3 replicas (1 primary and 2 slaves). Mongo version 5.0.4. Also SpringBoot WebFlux application with reactive mongodb driver. I often see duplications in database when I save documents using standard Spring Data Repository. For example: I have devices and commands for them. Each command has list of statuses.

Connection options:

spring:
  data:
    mongodb:
      uri: mongodb://${MONGO_USERNAME}:${MONGO_PASSWORD}@${MONGO_CLUSTER}/${DATABASE_NAME}?authSource=admin&readPreference=secondary&replicaSet=rs0&minPoolSize=20

CommandDocument:

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Document(collection = "command")
public class CommandDocument {

    @Id
    private String id;
    private String command;

    @DocumentReference
    private DeviceInfoDocument deviceInfo;
}

CommandStatusDocument:

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Document(collection = "commandStatus")
public class CommandStatusDocument {

    @Id
    private String id;
    private String commandId;
    private Instant timestamp;
    private CommandStatus commandStatus;

}

When I create command, I set status RECEIVED. When I send command to device I need to add status SENT_TO_DEVICE. The main chain of getting commands with statuses by device (very simplified):

@Override
public Flux<CommandGetResponse> getCommands(CommandGetRequest request) {
    return deviceInfoService.findByDeviceId(request.getDeviceId())
            .flatMap(commandService::findByDevice)
            .doOnNext(command ->
                    commandStatusService.createCommandStatusDocument(
                                  command,
                                  CommandStatus.SENT_TO_DEVICE))
                            .subscribe()
            )
            .flatMap(command -> command.getStatuses()
                    .map(statuses -> CommandGetResponse.from(command, statuses))
            );
}

Create a new commandStatusDocument:

@Override
public Mono<CommandStatusDocument> createCommandStatusDocument(CommandDocument commandDocument, CommandStatus status) {
    var commandStatus = CommandStatusDocument.builder()
            .commandStatus(status)
            .commandId(commandDocument.getId())
            .timestamp(Instant.now())
            .build();

    return commandStatusRepository.save(commandStatus);
}

Repository:

public interface CommandStatusRepository extends ReactiveMongoRepository<CommandStatusDocument, String> {}

When device gets list of commands I often see two statuses SENT_TO_DEVICE (with different IDs) for one command with small time difference. Like this:

db.commandStatus.find({commandId: "64d2b794216a5a4984bbf0ae"})
[
  {
    _id: ObjectId("64d2b794216a5a4984bbf0af"),
    commandId: '64d2b794216a5a4984bbf0ae',
    timestamp: ISODate("2023-08-08T21:45:56.283Z"),
    commandStatus: 'RECEIVED',
    _class: 'com.example.commandstatus.CommandStatusDocument'
  },
  {
    _id: ObjectId("64d2b7b85096f472ff71be83"),
    commandId: '64d2b794216a5a4984bbf0ae',
    timestamp: ISODate("2023-08-08T21:46:32.895Z"),
    commandStatus: 'SENT_TO_DEVICE',
    _class: 'com.example.commandstatus.CommandStatusDocument'
  },
  {
    _id: ObjectId("64d2b7b85096f472ff71be84"),
    commandId: '64d2b794216a5a4984bbf0ae',
    timestamp: ISODate("2023-08-08T21:46:32.899Z"),
    commandStatus: 'SENT_TO_DEVICE',
    _class: 'com.example.commandstatus.CommandStatusDocument'
  }
]

I tried to add check on existing status before save like this:

.doOnNext(command ->
    commandStatusService.findByStatus(command, CommandStatus.SENT_TO_DEVICE)                            .switchIfEmpty(commandStatusService.createCommandStatusDocument(
                                    command,
                                    CommandStatus.SENT_TO_DEVICE))
    .subscribe()
)

But it doesn't take any effect. What is possible reason of this behaviour? There are many places when it happens, not only for commands and statuses. Is it related to sort of replication things or my code?

Answer 1

public Flux<CommandGetResponse> getCommands(CommandGetRequest request) {
    return deviceInfoService.findByDeviceId(request.getDeviceId())
        .flatMap(commandService::findByDevice)
        .flatMap(command -> 
                commandStatusService.createCommandStatusDocument(
                              command,
                              CommandStatus.SENT_TO_DEVICE))
                  .map(doc -> Tuples.of(doc, command))
        )
        .flatMap(tuple -> {
              CommandDocument command = tuple.getT2();                   
              command.getStatuses()
                .map(statuses -> CommandGetResponse.from(command, statuses))
        });
}

Or you can probably move everything inside a single flatMap like:

public Flux<CommandGetResponse> getCommands(CommandGetRequest request) {
    return deviceInfoService.findByDeviceId(request.getDeviceId())
        .flatMap(commandService::findByDevice)
        .flatMap(command -> {
                Mono<CommandStatusDocument> status = commandStatusService.createCommandStatusDocument(
                              command,
                              CommandStatus.SENT_TO_DEVICE));
                Flux<CommandGetResponse> resp = command.getStatuses()
                    .map(statuses -> CommandGetResponse.from(command, statuses));
               return status.thenMany(resp);
        });
}