MongoTrackingToken.lowerBound() incompatible with ReplayToken.wasProcessedBeforeReset() — replays broken

[martijnvanderwoud]

Summary

After upgrading from Axon Framework 4.7.3 + axon-mongo 4.7.0 to Axon Framework 4.13.0 + axon-mongo 4.12.0, event processor replays are completely broken. ReplayToken.wasProcessedBeforeReset() returns false for events dated years before the reset token, causing ReplayToken.advancedTo() to incorrectly treat replayed events as "new events not seen before" instead of "events still well behind the reset position."

This is related to #9, which reported that MongoTrackingToken's trackedEvents grow unboundedly during replay. The current issue has a much more severe consequence: replays don't work at all, because the replay/non-replay classification is wrong for virtually every replayed event.

Root cause

ReplayToken.wasProcessedBeforeReset() was introduced in Axon Framework as a refactoring of the replay detection logic. It relies on lowerBound() + samePositionAs():

private static boolean wasProcessedBeforeReset(TrackingToken tokenAtReset, TrackingToken newToken) {
    TrackingToken resetLowerBound = WrappedToken.unwrapLowerBound(tokenAtReset);
    TrackingToken newTokenLowerBound = WrappedToken.unwrapLowerBound(newToken);
    TrackingToken resetTokenLowerNewToken = resetLowerBound.lowerBound(newTokenLowerBound);
    return resetTokenLowerNewToken.samePositionAs(newTokenLowerBound);
}

This works for GapAwareTrackingToken because its lowerBound() produces a token with a meaningful index, and its samePositionAs() compares indices.

MongoTrackingToken.lowerBound() computes the intersection of tracked event IDs:

public TrackingToken lowerBound(TrackingToken other) {
    MongoTrackingToken otherToken = (MongoTrackingToken) other;
    Map<String, Long> intersection = new HashMap<>(this.trackedEvents);
    trackedEvents.keySet().forEach(k -> {
        if (!otherToken.trackedEvents.containsKey(k)) {
            intersection.remove(k);
        }
    });
    return new MongoTrackingToken(min(timestamp, otherToken.timestamp), intersection);
}

During replay, the token at reset (recent) and the replayed token (old) have completely disjoint event IDs — they are from different time periods and the old events have long been trimmed from the reset token's lookback window. The intersection is empty.

The resulting degenerate MongoTrackingToken{timestamp=<min>, trackedEvents={}} then fails the default samePositionAs() check (which uses covers(other) && other.covers(this)), because covers() cannot verify the other token's events against an empty tracked events set — the fallback oldest-event threshold defaults to 0L.

Observed values during debug session

Variable	Value
tokenAtReset	MongoTrackingToken{timestamp=1776948514695, trackedEvents={a812a642-…=1776948514695}}
newToken	MongoTrackingToken{timestamp=1524402340572, trackedEvents={ef664838-…=1524402340176, 78bae6b9-…=1524402340572}}
resetLowerBound	same as tokenAtReset
newTokenLowerBound	same as newToken
resetTokenLowerNewToken	MongoTrackingToken{timestamp=1524402340572, trackedEvents={}} ← empty!
Result	false (should be true — newToken is ~8 years before tokenAtReset)

Versions

• Before (working): Axon Framework 4.7.3, axon-mongo 4.7.0
• After (broken): Axon Framework 4.13.0, axon-mongo 4.12.0

Suggested fix

MongoTrackingToken.lowerBound() should use the same "older than the oldest tracked event means trimmed from the lookback window, therefore definitely processed" reasoning that covers() already uses. Instead of a pure event ID intersection, include events from either token that the other token has necessarily processed — either because the event ID is in both tokens, or because the event predates the other token's lookback window:

@Override
public TrackingToken lowerBound(TrackingToken other) {
    Assert.isTrue(other instanceof MongoTrackingToken, () -> "Incompatible token type provided.");
    MongoTrackingToken otherToken = (MongoTrackingToken) other;

    long minTimestamp = min(timestamp, otherToken.timestamp);

    long thisOldest = this.trackedEvents.values().stream()
            .min(Comparator.naturalOrder()).orElse(minTimestamp);
    long otherOldest = otherToken.trackedEvents.values().stream()
            .min(Comparator.naturalOrder()).orElse(minTimestamp);

    Map<String, Long> events = new HashMap<>();

    this.trackedEvents.forEach((k, v) -> {
        if (otherToken.trackedEvents.containsKey(k) || v < otherOldest) {
            events.put(k, v);
        }
    });
    otherToken.trackedEvents.forEach((k, v) -> {
        if (this.trackedEvents.containsKey(k) || v < thisOldest) {
            events.putIfAbsent(k, v);
        }
    });

    return new MongoTrackingToken(minTimestamp, events);
}

This preserves the out-of-order delivery guarantee: events within the lookback window that only one token has seen are still excluded from the lower bound, so samePositionAs correctly returns false for genuinely unprocessed events near the boundary.

[martijnvanderwoud]

Update: the proposed lowerBound() fix is unsound

After implementing and testing the fix I proposed above, I found a correctness problem.

The flaw

The fix includes events from one token in the lower bound if their timestamp predates the other token's oldest tracked event (v < otherOldest). The assumption is: "if the event is older than my oldest tracked event, it must have fallen outside my lookback window, so I definitely already processed it."

This assumption doesn't hold. Consider:

• tokenA = MongoTrackingToken.of(time(1951), "a") — trackedEvents: {"a": 1951}
• tokenB = MongoTrackingToken.of(time(1950), "b").advanceTo(time(1951), "a", oneSecond).advanceTo(time(2000), "c", oneSecond) — trackedEvents: {"b": 1950, "a": 1951, "c": 2000}

This models a valid out-of-order scenario: event "b" at t=1950 arrived in the store after tokenA had already advanced to t=1951. tokenA has genuinely never processed "b". tokenA is not necessarily a new token — it could have been advancing for a long time, with all previous events trimmed because they fell outside the lookback window. There simply were no events between the previous one and "a" at t=1951.

Yet lowerBound() includes "b" because 1950 < tokenA's oldest tracked event (1951). The lower bound then incorrectly claims both tokens have processed "b", which could cause it to be skipped.

Root cause: the lookback duration is not stored in the token

The real "safe" boundary — the timestamp before which a token has definitely processed everything — is headTimestamp - lookbackDuration. But MongoTrackingToken discards the lookback duration after using it to trim in advanceTo(). Only the surviving tracked events remain, and the oldest one is merely an upper bound on the lookback window boundary, not the boundary itself. These can differ by nearly the entire lookback duration when events are sparse.

Without the lookback duration, we cannot distinguish between:

1. "This event predates my lookback window — I definitely processed it"
2. "This event is within my lookback window but I haven't received it yet (out-of-order arrival)"

Note that covers() uses the same heuristic and has the same limitation.

What would a correct fix require?

The token would need to store the effective lower bound of its lookback window (e.g., headTimestamp - lookbackDuration) so that lowerBound() and covers() can use the real boundary. However, this adds a field to a serialized token that is persisted in the database by tracking event processors, which has significant migration and compatibility implications.

[smcvb]

The token would need to store the effective lower bound of its lookback window (e.g., headTimestamp - lookbackDuration) so that lowerBound() and covers() can use the real boundary. However, this adds a field to a serialized token that is persisted in the database by tracking event processors, which has significant migration and compatibility implications.

However, given the severity, and live cycle of this extension (as pointed out here), I am leaning towards something "rough to make it work."
I'd prefer we leave (assuming we'd leave this extension) in a workable state for AF4 users until the end of time.

@martijnvanderwoud, would you be up for providing such a rough transition?
I'd wager we can add documentation to explain how users should move forward.
If, for once, that requires a manual adjustment of the tokens in your collections, than that's what it is.

[martijnvanderwoud]

Thanks for the quick reply, @smcvb! I see that @abuijze already submitted a PR for AxonIQ/AxonFramework#4453. That will bring us back to the situation where the unbounded growth of replay tokens only occurs if the tokenAtReset has no tracked events, which is rare, as described in #9.

So, hopefully, once Axon Framework, 4.13.1 is released, there will be no immediate need to fix MongoTrackingToken.lowerBound. That said, I think it would be good to have a correct definition of lowerBound() and covers() going forward. I think my client would be willing to let me spend some time on that, but it will probably depend on whether this extension will be supported for Axon Framework 5. We have several production applications using MongoDB for event storage and we need to decide whether to keep the current infrastructure or migrate to Axon Server

[smcvb]

Thanks for your feedback, @martijnvanderwoud! Happy to hear the immediate concern is alleviated. I'll see to it that we'll look for a date to release 4.13.1 with the ReplayToken fix soonish.

Given the Mongo Extension's life in AF5; I am afraid I am not sure yet. There are part that are valuable, and there are parts that aren't.
If you're looking for a quick way forward to move to AF5, migrating might be best. I'd expect the Mongo Extension to come quite late in our planning for this year. Sorry for the inconvenience this may cause to you and the team 🙏

[smcvb]

Adjusted the status and priority accordingly to the fixes made in AxonIQ/AxonFramework#4453. Keeping the issue open for the chance we need to revisit this once planning for this extension starts for AF5.