fix(amm): stop superseded context work

apps/amm/src/NewPositionRuntime.cpp

@@ -346,6 +346,7 @@ void NewPositionRuntime::contextAsync(const QVariantMap& request,
                                       bool refreshPublicData,
                                       ResultCallback callback)
 {
+    const quint64 contextGeneration = ++m_contextGeneration;
     if (network.status != QStringLiteral("ready")) {
         callback(contextState(network.status, network).toVariantMap());
         return;
@@ -367,17 +368,18 @@ void NewPositionRuntime::contextAsync(const QVariantMap& request,
         QStringLiteral("configId")).toString();
     QPointer<NewPositionRuntime> guard(this);
     m_sequencer->readAccounts({ configId }, refreshPublicData,
-        [guard, request, network, walletOpen, refreshPublicData,
+        [guard, request, network, walletOpen, refreshPublicData, contextGeneration,
          callback = std::move(callback)](QVector<WalletAccountRead> configReads) mutable {
-            if (!guard)
+            if (!guard || contextGeneration != guard->m_contextGeneration)
                 return;
             const QJsonObject config = accountReadJson(configReads.value(0));
             guard->m_sequencer->readAccounts(
                 guard->m_walletPublicAccountIds, refreshPublicData,
                 [guard, request, network, walletOpen, refreshPublicData, config,
+                 contextGeneration,
                  callback = std::move(callback)](
                     QVector<WalletAccountRead> walletReads) mutable {
-                    if (!guard)
+                    if (!guard || contextGeneration != guard->m_contextGeneration)
                         return;
                     const QJsonObject hints = QJsonObject::fromVariantMap(request);
                     QJsonArray configured;
@@ -415,11 +417,13 @@ void NewPositionRuntime::contextAsync(const QVariantMap& request,
                     }
                     guard->m_sequencer->readAccounts(definitionIds, refreshPublicData,
                         [guard, network, walletOpen, config, walletAccounts,
-                         configured, recent, resolved,
+                         configured, recent, resolved, contextGeneration,
                          callback = std::move(callback)](
                             QVector<WalletAccountRead> definitions) mutable {
-                            if (!guard)
+                            if (!guard
+                                || contextGeneration != guard->m_contextGeneration) {
                                 return;
+                            }
                             const AmmClientResult result = guard->m_client->context(QJsonObject {
                                 { QStringLiteral("networkId"), network.id },
                                 { QStringLiteral("networkFingerprint"), network.fingerprint },

apps/amm/src/NewPositionRuntime.h

@@ -103,6 +103,7 @@ private:
     FreshLpState m_freshLpState = FreshLpState::None;
     bool m_submitInFlight = false;
     quint64 m_walletGeneration = 0;
+    quint64 m_contextGeneration = 0;
     quint64 m_submitGeneration = 0;
     ResultCallback m_submitCallback;
 };

apps/amm/tests/cpp/NewPositionRuntimeTest.cpp

@@ -292,6 +292,7 @@ namespace {
 
         AmmClientResult tokenIds(const QJsonObject&) const override
         {
+            ++tokenIdsCalls;
             return success({ { QStringLiteral("status"), QStringLiteral("ok") } });
         }
 
@@ -313,6 +314,7 @@ namespace {
 
         AmmClientResult context(const QJsonObject&) const override
         {
+            ++contextCalls;
             return success({});
         }
 
@@ -381,6 +383,8 @@ namespace {
         QString quoteHash = QStringLiteral("sha256:expected");
         bool requiresFreshLp = true;
         mutable bool sawFreshLp = false;
+        mutable int tokenIdsCalls = 0;
+        mutable int contextCalls = 0;
         mutable int planCalls = 0;
         mutable int planFailuresRemaining = 0;
         mutable QStringList freshLpAccountIds;
@@ -702,6 +706,67 @@ int main(int argc, char** argv)
                 "matching endpoint should retain configured credentials"))
         return 1;
 
+    LocalRpcServer staleContextServer;
+    if (!expect(staleContextServer.listen(),
+                "stale-context sequencer should listen"))
+        return 1;
+    staleContextServer.holdResponses();
+    QTemporaryFile staleContextConfig;
+    if (!expect(staleContextConfig.open(),
+                "stale-context config should open"))
+        return 1;
+    staleContextConfig.write(QJsonDocument(QJsonObject {
+        { QStringLiteral("sequencer_addr"), staleContextServer.endpoint() },
+    }).toJson(QJsonDocument::Compact));
+    staleContextConfig.flush();
+    FakeWallet staleContextWallet;
+    FakeAmmClient staleContextClient;
+    SequencerClient staleContextSequencer(&staleContextClient);
+    if (!expect(staleContextSequencer.configure(staleContextConfig.fileName()),
+                "stale-context sequencer should configure"))
+        return 1;
+    NewPositionRuntime staleContextRuntime(
+        &staleContextWallet, &staleContextClient, &staleContextSequencer);
+    WalletAccount staleContextHolding;
+    staleContextHolding.address = QString(64, QLatin1Char('4'));
+    staleContextHolding.isPublic = true;
+    staleContextRuntime.setWalletAccounts({ staleContextHolding });
+    int staleContextCallbacks = 0;
+    int latestContextCallbacks = 0;
+    staleContextRuntime.contextAsync(
+        {}, readyNetwork(), true, false,
+        [&](QVariantMap) { ++staleContextCallbacks; });
+    if (!expect(waitForRequestCount(staleContextServer, 1),
+                "first context should begin the config read"))
+        return 1;
+    staleContextRuntime.contextAsync(
+        {}, readyNetwork(), true, false,
+        [&](QVariantMap) { ++latestContextCallbacks; });
+    QCoreApplication::processEvents(QEventLoop::AllEvents, 10);
+    if (!expect(staleContextServer.requestCount() == 1,
+                "superseding context should share the active config read"))
+        return 1;
+    staleContextServer.releaseNextResponse();
+    if (!expect(waitForRequestCount(staleContextServer, 2),
+                "latest context should continue with the wallet read"))
+        return 1;
+    if (!expect(staleContextCallbacks == 0
+                    && latestContextCallbacks == 0
+                    && staleContextClient.tokenIdsCalls == 0
+                    && staleContextClient.contextCalls == 0,
+                "context should wait for the latest wallet read"))
+        return 1;
+    staleContextServer.releaseNextResponse();
+    if (!expect(waitForCondition([&]() { return latestContextCallbacks == 1; }),
+                "latest context should complete"))
+        return 1;
+    if (!expect(staleContextCallbacks == 0
+                    && staleContextClient.tokenIdsCalls == 1
+                    && staleContextClient.contextCalls == 1
+                    && staleContextServer.requestCount() == 2,
+                "superseded context should stop before downstream work"))
+        return 1;
+
     LocalRpcServer server;
     if (!expect(server.listen(), "local sequencer should listen"))
         return 1;