Track superseded mempool errors separately

[tilacog]

Description

Mempools::execute() runs all configured mempools concurrently and returns the first one that succeeds. Previously, errors from mempools that lost the race were counted as real failures, even though the overall submission was successful. Dropped mempools were never recorded.

This skewed mempool counts by both:

• counting errors alongside a successful submission, and
• omitting counts for dropped mempools.

This PR keeps the racing behavior but changes how observation works.

Changes

Behavior

• On first success, emit mempool_succeeded for the winner and mempool_superseded for every other configured mempool.
• On all-failed, emit mempool_failed for each one of them.

Code specific

• Replace select_ok with FuturesUnordered in Mempools::execute so the consumer can observe each completion (crates/driver/src/domain/mempools.rs).
• Split observe::mempool_executed into mempool_succeeded(&SubmissionSuccess) and mempool_failed(&mempools::Error), dropping the Result<&S, &E> indirection now that each call site already knows which branch it is on. Behavior and emitted metrics are unchanged by the split.
• Add mempool_superseded(&Mempool, winner: &Mempool, &Settlement) which increments driver_mempool_submission with result="Superseded".

How to test

Existing driver unit tests cover the race semantics; this PR does not change the externally observable submission outcome, only how observation is sequenced and labeled. To verify manually:

1. Run the driver against a config with at least two mempools.
2. Trigger a settlement that succeeds via the public mempool.
3. Confirm Prometheus shows one result="Success" increment for the winner and one result="Superseded" increment for the loser; no Revert/Expired/Other from the loser.
4. Trigger a settlement that fails on every mempool and confirm each mempool gets its own non-Superseded failure label.

Alert query update needed when deploying

Per-mempool success counts both wins and races-lost (so happy and failure paths both emit N events for N configured mempools, keeping the ratio symmetric). Superseded stays as a separate label so dashboards can still distinguish wins from race-losses per mempool.

sum by (network) (increase(driver_mempool_submission{cow_fi_environment="prod",result=~"Success|Superseded"}[2h]))
/
sum by (network) (increase(driver_mempool_submission{cow_fi_environment="prod",result!="Disabled"}[2h])) < 0.6

[fleupold]

Is there a reason you are not using the PR template for the description?

I agree with the change, however I'd like to suggest that we interpret "superseeded" events as success wrt. how you envision to change the metric. A superseeded submission should be considered a successful one.

This way we receive N (# of mempool) events in the happy case, and N events in the failure case allowing us to keep our alert metric as a ratio of successful to failed ones (otherwise failed events would be weighted N times more than successful ones).

[tilacog]

Is there a reason you are not using the PR template for the description?

Apologies, I was in a rush and didn't account for that. I've updated the description to match the template.

I agree with the change, however I'd like to suggest that we interpret "superseeded" events as success wrt. how you envision to change the metric. A superseeded submission should be considered a successful one.

This way we receive N (# of mempool) events in the happy case, and N events in the failure case allowing us to keep our alert metric as a ratio of successful to failed ones (otherwise failed events would be weighted N times more than successful ones).

Agree. I've adjusted the suggested metric.

[gemini-code-assist]

Code Review

This pull request refactors the mempool execution logic to use FuturesUnordered, enabling more detailed tracking of success, failure, and superseded states. It also adds a blocks_passed method to the Error enum for improved block-level timing metrics. A high-severity logic error was identified where disabled mempools are incorrectly reported as 'Superseded' if another mempool wins the race, which would artificially inflate success rate metrics. A correction was suggested to preserve the 'Disabled' status during the racing process.

[MartinquaXD]

The idea to fix the metrics makes sense to me but the logic seems more complicated than necessary.

Wouldn't it also work to construct a hashmap mapping all mempools to a submission outcome?

1. initialize all mempools as superseeded
2. run new FuturesUnordered logic
3. match on each submit result and update the hashmap accordingly

At the end all submission futures that finished will have updated their own entry in the mapping and all futures that didn't finish were cancelled because some other pool was successful first so the original superseeded label is correct.

[tilacog]

I agree this got more complex than it should.

At the end all submission futures that finished will have updated their own entry in the mapping and all futures that didn't finish were cancelled because some other pool was successful first so the original superseeded label is correct.

If I'm reading this right, I think this arrangement still leaves us with a possible final state of [winner(1), errored(N), superseeded(M)] (out of [1+N+M] mempools).

On the hashmap point, I don't think it changes the design much. We still need to iterate over all mempools to set the disabled ones, then iterate by value to overwrite the errored ones with superseded on a success. That's basically the current design, just with a different container.

I'll try to simplify the current code a bit.

[tilacog]

I hope this is correct/appropriate, because deriving PartialEq on Mempool simplifies the filtering in the race post-processing.

Config carries several fields, most importantly the Mempool's Url and Name.

[MartinquaXD]

We still need to iterate over all mempools to set the disabled ones, then iterate by value to overwrite the errored ones with superseded on a success.

I think the difference is that the current approach spreads around the responsibilities of managing the metrics too much. With the approach suggested below there are 3 steps:

1. initialize counters
2. poll futures and update counters based on the results
3. interpret stats and update metrics

There is no need to pre-emptively partition out the disabled pools (also no need for a debug assert), no need to collect the errors into a vector to return the last one, and the logic of interpreting the results (interpreting errors as superseed when there is a success result) happens inside a singular function where the reasoning can be documented nicely.

        // initialize all pools with superseeded as that is the correct state when we
        // don't get to update the pools when one succeeds.
        let mut stats = self.mempools.iter().map(|mem| (&mem, Superseeded)).collect();

        let (submission, _remaining_futures) = select_ok(self.mempools.iter().map(|mempool| {
            async move {
                let result = self
                    .submit(mempool, settlement, submission_deadline, mode)
                    .instrument(tracing::info_span!("mempool", kind = mempool.to_string()))
                    .await;
                match &result {
                    Ok(_) => stats[&mempool] = Success,
                    Err(Disabled) => stats[&mempool] = Disabled,
                    Err(_) => stats[&mempool] = Error,
                }
                result
            }
            .boxed()
        }))
        .await?;

        // if stats.values().any(|val| val==success) we mark every error as superseeded
        // otherwise we recorde the given label
        self.update_metrics(stats);

        Ok(submission.tx_hash)

[tilacog]

Ok, here's what I landed on after addressing @MartinquaXD's comments:

I didn't use a Dashmap or Hashmap because hashing Mempool isn't straightforward (due to an float field in Config), so I kept a vector of pairs instead (Mempool, Outcome) and handled them by position.

I had to use FuturesUnordered instead of select_ok because sending submit's results out of the async closure wasn't possible due to lifetime constraints, plus mempool::Error isn't Clone since anyhow::Error isn't either... and I didn't want to add more complexity there.

So I followed the suggestions as closely as I could:

I used an Outcome enum to capture (and own) errors and submissions results by consuming the results from submit calls, and processed them in their metrics pipeline, finally recreating the returning result to avoid breaking the contract with execute callers (currently just Competition::process_settle_request). I added a few unit tests for that part since they were easy to write.

I think this is in good shape to merge. If we don't see any correctness issues, I'd recommend merging right away and improving the code later, because I think this will help with our current alerts on high mempool submission errors.