Transferprocess starts multiple times when transferring huge data

[Jens011203]

Bug Report

During testing to find out the amount of data I can transfer via the HTTP dataplane, I discovered that when transferring 1Gb, 4-6 Gb was received on consumer side.

Describe the Bug

When the data is too large and hence the transfer method takes a very long time, the process remains in the RECEIVED state when the performLogic is run again in the StateMachineManager.

private boolean processReceived(DataFlow dataFlow) {
        var request = dataFlow.toRequest();
        var transferService = transferServiceRegistry.resolveTransferService(request);

        if (transferService == null) {
            dataFlow.transitToFailed("No transferService available for DataFlow " + dataFlow.getId());
            update(dataFlow);
            return true;
        }

        return entityRetryProcessFactory.doAsyncProcess(dataFlow, () -> transferService.transfer(request))
                .entityRetrieve(id -> store.findById(id))
                .onSuccess((f, r) -> {
                    if (r.succeeded()) {
                        f.transitToCompleted();
                    } else {
                        f.transitToFailed(r.getFailureDetail());
                    }
                    update(f);
                })
                .onFailure((f, t) -> {
                    f.transitToReceived();
                    update(f);
                })
                .onRetryExhausted((f, t) -> {
                    f.transitToFailed(t.getMessage());
                    update(f);
                })
                .execute("start data flow");
    }

   @Override
    protected StateMachineManager.Builder configureStateMachineManager(StateMachineManager.Builder builder) {
        return builder
                .processor(processDataFlowInState(RECEIVED, this::processReceived))
                .processor(processDataFlowInState(COMPLETED, this::processCompleted))
                .processor(processDataFlowInState(FAILED, this::processFailed));
    }

    private Runnable loop() {
        return () -> {
            if (active.get()) {
                performLogic();
            }
        };
    }

    private void performLogic() {
        try {
            var processed = processors.stream()
                    .mapToLong(Processor::process)
                    .sum();

            waitStrategy.success();

            var delay = processed == 0 ? waitStrategy.waitForMillis() : 0;

            scheduleNextIterationIn(delay);
        } catch (Error e) {
            active.set(false);
            monitor.severe(format("StateMachineManager [%s] unrecoverable error", name), e);
        } catch (Throwable e) {
            monitor.severe(format("StateMachineManager [%s] error caught", name), e);
            scheduleNextIterationIn(waitStrategy.retryInMillis());
        }
    }

Expected Behavior

The same transferprocess should'nt be startet multiple times.

Steps to Reproduce

Steps to reproduce the behavior:

1. Transfer 1Gb over the http-data-plane

[github-actions]

Thanks for your contribution 🔥 We will take a look asap 🚀

[ndr-brt]

I also found out some issues with this design, I think this should be fixed by changing the state from RECEIVED to STARTED so then the data flow won't get picked up by the state machine again