Merge pull request #20902 from vespa-engine/toregge/prepare-for-throttling-replayv7.532.27

Prepare for throttling of replay.

16 files changed, 150 insertions, 44 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/common/CMakeLists.txt b/searchcore/src/vespa/searchcore/proton/common/CMakeLists.txt
index 421e602a9cf..03bad4c7eaf 100644
--- a/searchcore/src/vespa/searchcore/proton/common/CMakeLists.txt
+++ b/searchcore/src/vespa/searchcore/proton/common/CMakeLists.txt
@@ -18,6 +18,7 @@ vespa_add_library(searchcore_pcommon STATIC
     ipendinglidtracker.cpp
     operation_rate_tracker.cpp
     pendinglidtracker.cpp
+    replay_feedtoken_state.cpp
     select_utils.cpp
     selectcontext.cpp
     selectpruner.cpp
diff --git a/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp b/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp
index 675dce10be4..c74819577e9 100644
--- a/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp
+++ b/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp
@@ -33,6 +33,12 @@ State::setResult(ResultUP result, bool documentWasFound) {
     _result = std::move(result);
 }
 
+bool
+State::is_replay() const noexcept
+{
+    return false;
+}
+
 void
 State::fail()
 {
diff --git a/searchcore/src/vespa/searchcore/proton/common/feedtoken.h b/searchcore/src/vespa/searchcore/proton/common/feedtoken.h
index 8ccb4863878..53f551999d2 100644
--- a/searchcore/src/vespa/searchcore/proton/common/feedtoken.h
+++ b/searchcore/src/vespa/searchcore/proton/common/feedtoken.h
@@ -22,19 +22,33 @@ public:
     virtual void send(ResultUP result, bool documentWasFound) = 0;
 };
 
+
+/*
+ * Interface class for feed token state.
+ */
+class IState : public vespalib::IDestructorCallback {
+public:
+    virtual bool is_replay() const noexcept = 0;
+    virtual void fail() = 0;
+    virtual void setResult(ResultUP result, bool documentWasFound) = 0;
+    virtual const storage::spi::Result &getResult() = 0;
+};
+
+
 /**
  * This holds the result of the feed operation until it is either failed or acked.
  * Guarantees that the result is propagated back to the invoker via ITransport interface.
  */
-class State : public vespalib::IDestructorCallback {
+class State : public IState {
 public:
     State(const State &) = delete;
     State & operator = (const State &) = delete;
     State(ITransport & transport);
     ~State() override;
-    void fail();
-    void setResult(ResultUP result, bool documentWasFound);
-    const storage::spi::Result &getResult() { return *_result; }
+    bool is_replay() const noexcept override;
+    void fail() override;
+    void setResult(ResultUP result, bool documentWasFound) override;
+    const storage::spi::Result &getResult() override { return *_result; }
 protected:
     void ack();
 private:
@@ -71,7 +85,7 @@ make(std::shared_ptr<ITransport> transport) {
 
 }
 
-using FeedToken = std::shared_ptr<feedtoken::State>;
+using FeedToken = std::shared_ptr<feedtoken::IState>;
 
 } // namespace proton
 
diff --git a/searchcore/src/vespa/searchcore/proton/common/replay_feedtoken_state.cpp b/searchcore/src/vespa/searchcore/proton/common/replay_feedtoken_state.cpp
new file mode 100644
index 00000000000..a3a473c9548
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/common/replay_feedtoken_state.cpp
@@ -0,0 +1,37 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "replay_feedtoken_state.h"
+
+namespace proton::feedtoken {
+
+ReplayState::ReplayState(vespalib::SharedOperationThrottler::Token throttler_token)
+    : IState(),
+      _throttler_token(std::move(throttler_token))
+{
+}
+
+ReplayState::~ReplayState() = default;
+
+bool
+ReplayState::is_replay() const noexcept
+{
+    return true;
+}
+
+void
+ReplayState::fail()
+{
+}
+
+void
+ReplayState::setResult(ResultUP, bool)
+{
+}
+
+const storage::spi::Result&
+ReplayState::getResult()
+{
+    abort();
+}
+
+}
diff --git a/searchcore/src/vespa/searchcore/proton/common/replay_feedtoken_state.h b/searchcore/src/vespa/searchcore/proton/common/replay_feedtoken_state.h
new file mode 100644
index 00000000000..512f12a50af
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/common/replay_feedtoken_state.h
@@ -0,0 +1,26 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include "feedtoken.h"
+#include <vespa/vespalib/util/shared_operation_throttler.h>
+
+namespace proton::feedtoken {
+
+/*
+ * Feed token state used during replay. It contains a throttler token
+ * which allows the related shared operation throttler to track the completion
+ * of the feed operation.
+ */
+class ReplayState : public IState {
+    vespalib::SharedOperationThrottler::Token _throttler_token;
+public:
+    ~ReplayState() override;
+    ReplayState(vespalib::SharedOperationThrottler::Token throttler_token);
+    bool is_replay() const noexcept override;
+    void fail() override;
+    void setResult(ResultUP result, bool documentWasFound) override;
+    const storage::spi::Result &getResult() override;
+};
+
+}
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp b/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp
index d2626a0d9f4..1c8e17fc474 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp
@@ -8,8 +8,10 @@
 #include <vespa/searchcore/proton/bucketdb/ibucketdbhandler.h>
 #include <vespa/searchcore/proton/feedoperation/operations.h>
 #include <vespa/searchcore/proton/common/eventlogger.h>
+#include <vespa/searchcore/proton/common/replay_feedtoken_state.h>
 #include <vespa/vespalib/util/idestructorcallback.h>
 #include <vespa/vespalib/util/lambdatask.h>
+#include <vespa/vespalib/util/shared_operation_throttler.h>
 #include <cassert>
 
 #include <vespa/log/log.h>
@@ -69,14 +71,19 @@ public:
 
     ~TransactionLogReplayPacketHandler() override = default;
 
+    FeedToken make_replay_feed_token() {
+        vespalib::SharedOperationThrottler::Token throttler_token;
+        return std::make_shared<feedtoken::ReplayState>(std::move(throttler_token));
+    }
+
     void replay(const PutOperation &op) override {
-        _feed_view_ptr->handlePut(FeedToken(), op);
+        _feed_view_ptr->handlePut(make_replay_feed_token(), op);
     }
     void replay(const RemoveOperation &op) override {
-        _feed_view_ptr->handleRemove(FeedToken(), op);
+        _feed_view_ptr->handleRemove(make_replay_feed_token(), op);
     }
     void replay(const UpdateOperation &op) override {
-        _feed_view_ptr->handleUpdate(FeedToken(), op);
+        _feed_view_ptr->handleUpdate(make_replay_feed_token(), op);
     }
     void replay(const NoopOperation &) override {} // ignored
     void replay(const NewConfigOperation &op) override {
@@ -84,7 +91,7 @@ public:
     }
 
     void replay(const DeleteBucketOperation &op) override {
-        _feed_view_ptr->handleDeleteBucket(op, IDestructorCallback::SP());
+        _feed_view_ptr->handleDeleteBucket(op, make_replay_feed_token());
     }
     void replay(const SplitBucketOperation &op) override {
         _bucketDBHandler.handleSplit(op.getSerialNum(), op.getSource(),
@@ -95,15 +102,15 @@ public:
                                     op.getSource2(), op.getTarget());
     }
     void replay(const PruneRemovedDocumentsOperation &op) override {
-        _feed_view_ptr->handlePruneRemovedDocuments(op, IDestructorCallback::SP());
+        _feed_view_ptr->handlePruneRemovedDocuments(op, make_replay_feed_token());
     }
     void replay(const MoveOperation &op) override {
-        _feed_view_ptr->handleMove(op, IDestructorCallback::SP());
+        _feed_view_ptr->handleMove(op, make_replay_feed_token());
     }
     void replay(const CreateBucketOperation &) override {
     }
     void replay(const CompactLidSpaceOperation &op) override {
-        _feed_view_ptr->handleCompactLidSpace(op, IDestructorCallback::SP());
+        _feed_view_ptr->handleCompactLidSpace(op, make_replay_feed_token());
     }
     NewConfigOperation::IStreamHandler &getNewConfigStreamHandler() override {
         return _config_store;
diff --git a/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.cpp b/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.cpp
index 53b12972c44..9b1c6de15df 100644
--- a/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.cpp
@@ -1,14 +1,23 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "operationdonecontext.h"
+#include <vespa/searchcore/proton/common/feedtoken.h>
 
 namespace proton {
 
-OperationDoneContext::OperationDoneContext(IDestructorCallback::SP token)
-    : _token(std::move(token))
+OperationDoneContext::OperationDoneContext(std::shared_ptr<feedtoken::IState> token, std::shared_ptr<vespalib::IDestructorCallback> done_callback)
+    : _token(std::move(token)),
+      _done_callback(std::move(done_callback))
 {
 }
 
 OperationDoneContext::~OperationDoneContext() = default;
 
+bool
+OperationDoneContext::is_replay() const
+{
+    return (!_token || _token->is_replay());
+}
+
+
 }  // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.h b/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.h
index 113aae8c9bf..baf0e98d5a4 100644
--- a/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.h
+++ b/searchcore/src/vespa/searchcore/proton/server/operationdonecontext.h
@@ -4,6 +4,8 @@
 
 #include <vespa/vespalib/util/idestructorcallback.h>
 
+namespace proton::feedtoken { class IState; }
+
 namespace proton {
 
 /**
@@ -16,13 +18,13 @@ namespace proton {
 class OperationDoneContext : public vespalib::IDestructorCallback
 {
 public:
-    using IDestructorCallback = vespalib::IDestructorCallback;
-    OperationDoneContext(IDestructorCallback::SP token);
+    OperationDoneContext(std::shared_ptr<feedtoken::IState> token, std::shared_ptr<IDestructorCallback> done_callback);
 
     ~OperationDoneContext() override;
-    bool hasToken() const { return static_cast<bool>(_token); }
+    bool is_replay() const;
 private:
-    IDestructorCallback::SP _token;
+    std::shared_ptr<feedtoken::IState> _token;
+    std::shared_ptr<vespalib::IDestructorCallback> _done_callback;
 };
 
 }  // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/server/putdonecontext.cpp b/searchcore/src/vespa/searchcore/proton/server/putdonecontext.cpp
index 7801f21c906..695f39c0097 100644
--- a/searchcore/src/vespa/searchcore/proton/server/putdonecontext.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/putdonecontext.cpp
@@ -9,9 +9,11 @@ using document::Document;
 
 namespace proton {
 
-PutDoneContext::PutDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted,
+PutDoneContext::PutDoneContext(std::shared_ptr<feedtoken::IState> token,
+                               std::shared_ptr<vespalib::IDestructorCallback> done_callback,
+                               IPendingLidTracker::Token uncommitted,
                                std::shared_ptr<const Document> doc, uint32_t lid)
-    : OperationDoneContext(std::move(token)),
+    : OperationDoneContext(std::move(token), std::move(done_callback)),
       _uncommitted(std::move(uncommitted)),
       _lid(lid),
       _docIdLimit(nullptr),
diff --git a/searchcore/src/vespa/searchcore/proton/server/putdonecontext.h b/searchcore/src/vespa/searchcore/proton/server/putdonecontext.h
index 00e8ac46c93..e0f55816314 100644
--- a/searchcore/src/vespa/searchcore/proton/server/putdonecontext.h
+++ b/searchcore/src/vespa/searchcore/proton/server/putdonecontext.h
@@ -28,7 +28,9 @@ class PutDoneContext : public OperationDoneContext
     std::shared_ptr<const document::Document> _doc;
 
 public:
-    PutDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted,
+    PutDoneContext(std::shared_ptr<feedtoken::IState> token,
+                   std::shared_ptr<vespalib::IDestructorCallback> done_callback,
+                   IPendingLidTracker::Token uncommitted,
                    std::shared_ptr<const document::Document> doc, uint32_t lid);
     ~PutDoneContext() override;
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/removedonecontext.cpp b/searchcore/src/vespa/searchcore/proton/server/removedonecontext.cpp
index 558a57bb8c6..0c93992d427 100644
--- a/searchcore/src/vespa/searchcore/proton/server/removedonecontext.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/removedonecontext.cpp
@@ -4,8 +4,8 @@
 
 namespace proton {
 
-RemoveDoneContext::RemoveDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted)
-    : OperationDoneContext(std::move(token)),
+RemoveDoneContext::RemoveDoneContext(std::shared_ptr<feedtoken::IState> token, std::shared_ptr<IDestructorCallback> done_callback, IPendingLidTracker::Token uncommitted)
+    : OperationDoneContext(std::move(token), std::move(done_callback)),
       _uncommitted(std::move(uncommitted))
 {
 }
diff --git a/searchcore/src/vespa/searchcore/proton/server/removedonecontext.h b/searchcore/src/vespa/searchcore/proton/server/removedonecontext.h
index 28e15389bb2..62db0f20b84 100644
--- a/searchcore/src/vespa/searchcore/proton/server/removedonecontext.h
+++ b/searchcore/src/vespa/searchcore/proton/server/removedonecontext.h
@@ -19,7 +19,7 @@ class RemoveDoneContext : public OperationDoneContext
     IPendingLidTracker::Token _uncommitted;
 
 public:
-    RemoveDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted);
+    RemoveDoneContext(std::shared_ptr<feedtoken::IState>, std::shared_ptr<vespalib::IDestructorCallback> done_callback, IPendingLidTracker::Token uncommitted);
     ~RemoveDoneContext() override;
 };
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp
index 97bd940b403..3f1d290acf3 100644
--- a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp
@@ -43,10 +43,10 @@ namespace proton {
 namespace {
 
 std::shared_ptr<PutDoneContext>
-createPutDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted,
+createPutDoneContext(FeedToken token, std::shared_ptr<IDestructorCallback> done_callback, IPendingLidTracker::Token uncommitted,
                      std::shared_ptr<const Document> doc, uint32_t lid)
 {
-    return std::make_shared<PutDoneContext>(std::move(token), std::move(uncommitted), std::move(doc), lid);
+    return std::make_shared<PutDoneContext>(std::move(token), std::move(done_callback), std::move(uncommitted), std::move(doc), lid);
 }
 
 std::shared_ptr<UpdateDoneContext>
@@ -66,9 +66,9 @@ void setPrev(DocumentOperation &op, const documentmetastore::IStore::Result &res
 }
 
 std::shared_ptr<RemoveDoneContext>
-createRemoveDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted)
+createRemoveDoneContext(FeedToken token, std::shared_ptr<IDestructorCallback> done_callback, IPendingLidTracker::Token uncommitted)
 {
-    return std::make_shared<RemoveDoneContext>(std::move(token), std::move(uncommitted));
+    return std::make_shared<RemoveDoneContext>(std::move(token), std::move(done_callback), std::move(uncommitted));
 }
 
 class SummaryPutDoneContext : public OperationDoneContext
@@ -80,7 +80,7 @@ public:
 };
 
 SummaryPutDoneContext::SummaryPutDoneContext(FeedToken token, IPendingLidTracker::Token uncommitted)
-    : OperationDoneContext(std::move(token)),
+    : OperationDoneContext(std::move(token), {}),
       _uncommitted(std::move(uncommitted))
 {}
 
@@ -249,7 +249,7 @@ StoreOnlyFeedView::internalPut(FeedToken token, const PutOperation &putOp)
         if (putOp.changedDbdId() && useDocumentMetaStore(serialNum)) {
             _gidToLidChangeHandler.notifyPut(token, docId.getGlobalId(), putOp.getLid(), serialNum);
         }
-        auto onWriteDone = createPutDoneContext(std::move(token), get_pending_lid_token(putOp), doc, putOp.getLid());
+        auto onWriteDone = createPutDoneContext(std::move(token), {}, get_pending_lid_token(putOp), doc, putOp.getLid());
         putSummary(serialNum, putOp.getLid(), doc, onWriteDone);
         putAttributes(serialNum, putOp.getLid(), *doc, onWriteDone);
         putIndexedFields(serialNum, putOp.getLid(), doc, onWriteDone);
@@ -257,7 +257,7 @@ StoreOnlyFeedView::internalPut(FeedToken token, const PutOperation &putOp)
     if (docAlreadyExists && putOp.changedDbdId()) {
         //TODO, better to have an else than an assert ?
         assert(!putOp.getValidDbdId(_params._subDbId));
-        internalRemove(std::move(token), _pendingLidsForCommit->produce(putOp.getPrevLid()), serialNum, putOp.getPrevLid());
+        internalRemove(std::move(token), {}, _pendingLidsForCommit->produce(putOp.getPrevLid()), serialNum, putOp.getPrevLid());
     }
 }
 
@@ -458,13 +458,13 @@ StoreOnlyFeedView::makeUpdatedDocument(bool useDocStore, Lid lid, const Document
     Document::UP prevDoc = _summaryAdapter->get(lid, *_repo);
     Document::UP newDoc;
     vespalib::nbostream newStream(12345);
-    assert(!onWriteDone->hasToken() || useDocStore);
+    assert(onWriteDone->is_replay() || useDocStore);
     if (useDocStore) {
         assert(prevDoc);
     }
     if (!prevDoc) {
         // Replaying, document removed later before summary was flushed.
-        assert(!onWriteDone->hasToken());
+        assert(onWriteDone->is_replay());
         // If we've passed serial number for flushed index then we could
         // also check that this operation is marked for ignore by index
         // proxy.
@@ -478,7 +478,7 @@ StoreOnlyFeedView::makeUpdatedDocument(bool useDocStore, Lid lid, const Document
         } else {
             // Replaying, document removed and lid reused before summary
             // was flushed.
-            assert(!onWriteDone->hasToken() && !useDocStore);
+            assert(onWriteDone->is_replay() && !useDocStore);
         }
     }
     promisedDoc.set_value(std::move(newDoc));
@@ -552,7 +552,7 @@ StoreOnlyFeedView::internalRemove(FeedToken token, const RemoveOperationWithDocI
         if (rmOp.changedDbdId()) {
             //TODO Prefer else over assert ?
             assert(!rmOp.getValidDbdId(_params._subDbId));
-            internalRemove(std::move(token), _pendingLidsForCommit->produce(rmOp.getPrevLid()), serialNum, rmOp.getPrevLid());
+            internalRemove(std::move(token), {}, _pendingLidsForCommit->produce(rmOp.getPrevLid()), serialNum, rmOp.getPrevLid());
         }
     }
 }
@@ -569,16 +569,16 @@ StoreOnlyFeedView::internalRemove(FeedToken token, const RemoveOperationWithGid
     if (rmOp.getValidPrevDbdId(_params._subDbId)) {
         if (rmOp.changedDbdId()) {
             assert(!rmOp.getValidDbdId(_params._subDbId));
-            internalRemove(std::move(token), _pendingLidsForCommit->produce(rmOp.getPrevLid()), serialNum, rmOp.getPrevLid());
+            internalRemove(std::move(token), {}, _pendingLidsForCommit->produce(rmOp.getPrevLid()), serialNum, rmOp.getPrevLid());
         }
     }
 }
 
 void
-StoreOnlyFeedView::internalRemove(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted, SerialNum serialNum, Lid lid)
+StoreOnlyFeedView::internalRemove(FeedToken token, std::shared_ptr<IDestructorCallback> done_callback, IPendingLidTracker::Token uncommitted, SerialNum serialNum, Lid lid)
 {
     _lidReuseDelayer.delayReuse(lid);
-    auto onWriteDone = createRemoveDoneContext(std::move(token), std::move(uncommitted));
+    auto onWriteDone = createRemoveDoneContext(std::move(token), std::move(done_callback), std::move(uncommitted));
     removeSummary(serialNum, lid, onWriteDone);
     removeAttributes(serialNum, lid, onWriteDone);
     removeIndexedFields(serialNum, lid, onWriteDone);
@@ -706,13 +706,13 @@ StoreOnlyFeedView::handleMove(const MoveOperation &moveOp, DoneCallback doneCtx)
         if (moveOp.changedDbdId() && useDocumentMetaStore(serialNum)) {
             _gidToLidChangeHandler.notifyPut(FeedToken(), docId.getGlobalId(), moveOp.getLid(), serialNum);
         }
-        auto onWriteDone = createPutDoneContext(doneCtx, _pendingLidsForCommit->produce(moveOp.getLid()), doc, moveOp.getLid());
+        auto onWriteDone = createPutDoneContext({}, doneCtx, _pendingLidsForCommit->produce(moveOp.getLid()), doc, moveOp.getLid());
         putSummary(serialNum, moveOp.getLid(), doc, onWriteDone);
         putAttributes(serialNum, moveOp.getLid(), *doc, onWriteDone);
         putIndexedFields(serialNum, moveOp.getLid(), doc, onWriteDone);
     }
     if (docAlreadyExists && moveOp.changedDbdId()) {
-        internalRemove(doneCtx, _pendingLidsForCommit->produce(moveOp.getPrevLid()), serialNum, moveOp.getPrevLid());
+        internalRemove({}, doneCtx, _pendingLidsForCommit->produce(moveOp.getPrevLid()), serialNum, moveOp.getPrevLid());
     }
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h
index c25accaf4a4..bd8509fa796 100644
--- a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h
+++ b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h
@@ -180,7 +180,7 @@ private:
     // returns the number of documents removed.
     size_t removeDocuments(const RemoveDocumentsOperation &op, bool remove_index_and_attribute_fields, DoneCallback onDone);
 
-    void internalRemove(IDestructorCallbackSP token, IPendingLidTracker::Token uncommitted, SerialNum serialNum, Lid lid);
+    void internalRemove(FeedToken token, std::shared_ptr<vespalib::IDestructorCallback> done_callback,IPendingLidTracker::Token uncommitted, SerialNum serialNum, Lid lid);
 
     IPendingLidTracker::Token get_pending_lid_token(const DocumentOperation &op);
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.cpp b/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.cpp
index 5842965a18a..16099700980 100644
--- a/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.cpp
@@ -7,8 +7,8 @@ using document::Document;
 
 namespace proton {
 
-UpdateDoneContext::UpdateDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted, const document::DocumentUpdate::SP &upd)
-    : OperationDoneContext(std::move(token)),
+UpdateDoneContext::UpdateDoneContext(std::shared_ptr<feedtoken::IState> token, IPendingLidTracker::Token uncommitted, const document::DocumentUpdate::SP &upd)
+    : OperationDoneContext(std::move(token), {}),
       _uncommitted(std::move(uncommitted)),
       _upd(upd),
       _doc()
diff --git a/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.h b/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.h
index 62448102f30..7dd6c4a61f2 100644
--- a/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.h
+++ b/searchcore/src/vespa/searchcore/proton/server/updatedonecontext.h
@@ -24,7 +24,7 @@ class UpdateDoneContext : public OperationDoneContext
     document::DocumentUpdate::SP _upd;
     std::shared_future<std::unique_ptr<const document::Document>> _doc;
 public:
-    UpdateDoneContext(IDestructorCallback::SP token, IPendingLidTracker::Token uncommitted, const document::DocumentUpdate::SP &upd);
+    UpdateDoneContext(std::shared_ptr<feedtoken::IState> token, IPendingLidTracker::Token uncommitted, const document::DocumentUpdate::SP &upd);
     ~UpdateDoneContext() override;
 
     const document::DocumentUpdate &getUpdate() { return *_upd; }