Merge pull request #20288 from vespa-engine/balder/add-donecallback-token-to-remove-documents

Add DoneCallback to removeDocuments.

14 files changed, 72 insertions, 79 deletions

diff --git a/searchcore/src/tests/proton/documentdb/combiningfeedview/combiningfeedview_test.cpp b/searchcore/src/tests/proton/documentdb/combiningfeedview/combiningfeedview_test.cpp
index dc51d0df8a1..a57787d417e 100644
--- a/searchcore/src/tests/proton/documentdb/combiningfeedview/combiningfeedview_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/combiningfeedview/combiningfeedview_test.cpp
@@ -23,18 +23,10 @@ using namespace proton;
 
 typedef std::vector<IFeedView::SP> FeedViewVector;
 
-struct MyStreamHandler : public NewConfigOperation::IStreamHandler
-{
-    void serializeConfig(SerialNum, vespalib::nbostream &) override {}
-    void deserializeConfig(SerialNum, vespalib::nbostream &) override {}
-};
-
-
 struct MyFeedView : public test::DummyFeedView
 {
     typedef std::shared_ptr<MyFeedView> SP;
     DocumentMetaStore    _metaStore;
-    MyStreamHandler      _streamHandler;
     uint32_t             _preparePut;
     uint32_t             _handlePut;
     uint32_t             _prepareRemove;
@@ -56,7 +48,6 @@ struct MyFeedView : public test::DummyFeedView
                    DocumentMetaStore::getFixedName(),
                    search::GrowStrategy(),
                    subDbType),
-        _streamHandler(),
         _preparePut(0),
         _handlePut(0),
         _prepareRemove(0),
@@ -82,11 +73,11 @@ struct MyFeedView : public test::DummyFeedView
     void prepareRemove(RemoveOperation &) override { ++_prepareRemove; }
     void handleRemove(FeedToken, const RemoveOperation &) override { ++_handleRemove; }
     void prepareDeleteBucket(DeleteBucketOperation &) override { ++_prepareDeleteBucket; }
-    void handleDeleteBucket(const DeleteBucketOperation &) override { ++_handleDeleteBucket; }
+    void handleDeleteBucket(const DeleteBucketOperation &, DoneCallback) override { ++_handleDeleteBucket; }
     void prepareMove(MoveOperation &) override { ++_prepareMove; }
-    void handleMove(const MoveOperation &, IDestructorCallback::SP) override { ++_handleMove; }
+    void handleMove(const MoveOperation &, DoneCallback) override { ++_handleMove; }
     void heartBeat(SerialNum, DoneCallback) override { ++_heartBeat; }
-    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &) override { ++_handlePrune; }
+    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &, DoneCallback) override { ++_handlePrune; }
     void handleCompactLidSpace(const CompactLidSpaceOperation &op, DoneCallback) override {
         _wantedLidLimit = op.getLidLimit();
     }
@@ -362,7 +353,7 @@ TEST_F("require that delete bucket is sent to all feed views", Fixture)
     EXPECT_EQUAL(1u, f._ready._view->_prepareDeleteBucket);
     EXPECT_EQUAL(1u, f._removed._view->_prepareDeleteBucket);
     EXPECT_EQUAL(1u, f._notReady._view->_prepareDeleteBucket);
-    f._view.handleDeleteBucket(op);
+    f._view.handleDeleteBucket(op, IDestructorCallback::SP());
     EXPECT_EQUAL(1u, f._ready._view->_handleDeleteBucket);
     EXPECT_EQUAL(1u, f._removed._view->_handleDeleteBucket);
     EXPECT_EQUAL(1u, f._notReady._view->_handleDeleteBucket);
@@ -381,7 +372,7 @@ TEST_F("require that heart beat is sent to all feed views", Fixture)
 TEST_F("require that prune removed documents is sent to removed view", Fixture)
 {
     PruneRemovedDocumentsOperation op;
-    f._view.handlePruneRemovedDocuments(op);
+    f._view.handlePruneRemovedDocuments(op, IDestructorCallback::SP());
     EXPECT_EQUAL(0u, f._ready._view->_handlePrune);
     EXPECT_EQUAL(1u, f._removed._view->_handlePrune);
     EXPECT_EQUAL(0u, f._notReady._view->_handlePrune);
diff --git a/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp b/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp
index 1963e5e9dbc..904937a26da 100644
--- a/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/feedhandler/feedhandler_test.cpp
@@ -236,7 +236,7 @@ struct MyFeedView : public test::DummyFeedView {
     }
     void handleMove(const MoveOperation &, DoneCallback) override { ++move_count; }
     void heartBeat(SerialNum, DoneCallback) override { ++heartbeat_count; }
-    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &) override { ++prune_removed_count; }
+    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &, DoneCallback) override { ++prune_removed_count; }
     const ISimpleDocumentMetaStore *getDocumentMetaStorePtr() const override {
         return nullptr;
     }
diff --git a/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp b/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp
index 4c566f6053b..21daba2471d 100644
--- a/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/feedview/feedview_test.cpp
@@ -515,11 +515,6 @@ struct FixtureBase
     }
 
     template <typename FunctionType>
-    void runInMasterAndSyncAll(FunctionType func) {
-        test::runInMaster(_writeService, func);
-        _writeServiceReal.sync_all_executors();
-    }
-    template <typename FunctionType>
     void runInMaster(FunctionType func) {
         test::runInMaster(_writeService, func);
     }
@@ -617,10 +612,10 @@ struct FixtureBase
         gate.await();
     }
 
-    void performDeleteBucket(DeleteBucketOperation &op) {
+    void performDeleteBucket(DeleteBucketOperation &op, IDestructorCallback::SP onDone) {
         getFeedView().prepareDeleteBucket(op);
         op.setSerialNum(++serial);
-        getFeedView().handleDeleteBucket(op);
+        getFeedView().handleDeleteBucket(op, onDone);
     }
 
     void performForceCommit(IDestructorCallback::SP onDone) {
@@ -656,7 +651,7 @@ struct FixtureBase
         auto &fv = getFeedView();
         CompactLidSpaceOperation op(0, wantedLidLimit);
         op.setSerialNum(++serial);
-        fv.handleCompactLidSpace(op, std::move(onDone));
+        fv.handleCompactLidSpace(op, onDone);
     }
     void compactLidSpaceAndWait(uint32_t wantedLidLimit) {
         Gate gate;
@@ -950,7 +945,11 @@ TEST_F("require that handleDeleteBucket() removes documents", SearchableFeedView
 
     // delete bucket for user 1
     DeleteBucketOperation op(docs[0].bid);
-    f.runInMasterAndSyncAll([&]() { f.performDeleteBucket(op); });
+    vespalib::Gate gate;
+    f.runInMaster([&, onDone=std::make_shared<GateCallback>(gate)]() {
+        f.performDeleteBucket(op, std::move(onDone));
+    });
+    gate.await();
     f.dms_commit();
 
     EXPECT_EQUAL(0u, f.getBucketDB()->get(docs[0].bid).getDocumentCount());
diff --git a/searchcore/src/tests/proton/documentdb/storeonlyfeedview/storeonlyfeedview_test.cpp b/searchcore/src/tests/proton/documentdb/storeonlyfeedview/storeonlyfeedview_test.cpp
index fb609b1829e..1c2d903fead 100644
--- a/searchcore/src/tests/proton/documentdb/storeonlyfeedview/storeonlyfeedview_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/storeonlyfeedview/storeonlyfeedview_test.cpp
@@ -114,11 +114,11 @@ struct MyMinimalFeedView : public MyMinimalFeedViewBase, public StoreOnlyFeedVie
         StoreOnlyFeedView::removeIndexedFields(s, l, onWriteDone);
         ++removeMultiIndexFieldsCount;
     }
-    void heartBeatIndexedFields(SerialNum s, IDestructorCallbackSP onDone) override {
+    void heartBeatIndexedFields(SerialNum s, DoneCallback onDone) override {
         StoreOnlyFeedView::heartBeatIndexedFields(s, onDone);
         ++heartBeatIndexedFieldsCount;
     }
-    void heartBeatAttributes(SerialNum s, IDestructorCallbackSP onDone) override {
+    void heartBeatAttributes(SerialNum s, DoneCallback onDone) override {
         StoreOnlyFeedView::heartBeatAttributes(s, onDone);
         ++heartBeatAttributesCount;
     }
@@ -383,7 +383,11 @@ TEST_F("require that prune removed documents removes documents",
     PruneRemovedDocumentsOperation op(lids->getDocIdLimit(), subdb_id);
     op.setLidsToRemove(lids);
     op.setSerialNum(1);  // allows use of meta store.
-    f.runInMasterAndSyncAll([&]() { f.feedview->handlePruneRemovedDocuments(op); });
+    vespalib::Gate gate;
+    f.runInMaster([&, onDone=std::make_shared<vespalib::GateCallback>(gate)]() {
+        f.feedview->handlePruneRemovedDocuments(op, std::move(onDone));
+    });
+    gate.await();
 
     EXPECT_EQUAL(2, f.removeCount);
     EXPECT_FALSE(f.metaStore->validLid(1));
diff --git a/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.cpp b/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.cpp
index 23ca17c7656..297a9b9254f 100644
--- a/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.cpp
@@ -187,10 +187,10 @@ CombiningFeedView::prepareDeleteBucket(DeleteBucketOperation &delOp)
 }
 
 void
-CombiningFeedView::handleDeleteBucket(const DeleteBucketOperation &delOp)
+CombiningFeedView::handleDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone)
 {
     for (const auto &view : _views) {
-        view->handleDeleteBucket(delOp);
+        view->handleDeleteBucket(delOp, onDone);
     }
 }
 
@@ -203,7 +203,7 @@ CombiningFeedView::prepareMove(MoveOperation &moveOp)
 }
 
 void
-CombiningFeedView::handleMove(const MoveOperation &moveOp, IDestructorCallback::SP moveDoneCtx)
+CombiningFeedView::handleMove(const MoveOperation &moveOp, DoneCallback moveDoneCtx)
 {
     assert(moveOp.getValidDbdId());
     uint32_t subDbId = moveOp.getSubDbId();
@@ -235,9 +235,9 @@ CombiningFeedView::forceCommit(const CommitParam & param, DoneCallback onDone)
 
 void
 CombiningFeedView::
-handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp)
+handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp, DoneCallback onDone)
 {
-    getRemFeedView()->handlePruneRemovedDocuments(pruneOp);
+    getRemFeedView()->handlePruneRemovedDocuments(pruneOp, onDone);
 }
 
 void
@@ -245,7 +245,7 @@ CombiningFeedView::handleCompactLidSpace(const CompactLidSpaceOperation &op, Don
 {
     uint32_t subDbId = op.getSubDbId();
     assert(subDbId < _views.size());
-    _views[subDbId]->handleCompactLidSpace(op, std::move(onDone));
+    _views[subDbId]->handleCompactLidSpace(op, onDone);
 }
 
 void
diff --git a/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h b/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h
index 81a4f4f9588..ff640a32887 100644
--- a/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h
+++ b/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h
@@ -72,11 +72,11 @@ public:
     void prepareRemove(RemoveOperation &rmOp) override;
     void handleRemove(FeedToken token, const RemoveOperation &rmOp) override;
     void prepareDeleteBucket(DeleteBucketOperation &delOp) override;
-    void handleDeleteBucket(const DeleteBucketOperation &delOp) override;
     void prepareMove(MoveOperation &putOp) override;
-    void handleMove(const MoveOperation &moveOp, std::shared_ptr<vespalib::IDestructorCallback> moveDoneCtx) override;
+    void handleDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone) override;
+    void handleMove(const MoveOperation &moveOp, DoneCallback onDone) override;
     void heartBeat(search::SerialNum serialNum, DoneCallback onDone) override;
-    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp) override;
+    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp, DoneCallback onDone) override;
     void handleCompactLidSpace(const CompactLidSpaceOperation &op, DoneCallback onDone) override;
 
     // Called by document db executor
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp b/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp
index dc49aeb8f88..0ac91870eab 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp
@@ -281,7 +281,7 @@ FeedHandler::performDeleteBucket(FeedToken token, DeleteBucketOperation &op) {
     _activeFeedView->prepareDeleteBucket(op);
     appendOperation(op, token);
     // Delete documents in bucket
-    _activeFeedView->handleDeleteBucket(op);
+    _activeFeedView->handleDeleteBucket(op, token);
     // Delete bucket itself, should no longer have documents.
     _bucketDBHandler->handleDeleteBucket(op.getBucketId());
 
@@ -375,7 +375,9 @@ FeedHandler::changeFeedState(FeedStateSP newState)
     if (_writeService.master().isCurrentThread()) {
         doChangeFeedState(std::move(newState));
     } else {
-        _writeService.master().execute(makeLambdaTask([this, newState=std::move(newState)] () { doChangeFeedState(std::move(newState));}));
+        _writeService.master().execute(makeLambdaTask([this, newState=std::move(newState)] () {
+            doChangeFeedState(std::move(newState));
+        }));
         _writeService.master().sync();
     }
 }
@@ -793,13 +795,13 @@ FeedHandler::eof()
 }
 
 void
-FeedHandler::
-performPruneRemovedDocuments(PruneRemovedDocumentsOperation &pruneOp)
+FeedHandler::performPruneRemovedDocuments(PruneRemovedDocumentsOperation &pruneOp)
 {
     const LidVectorContext::SP lids_to_remove = pruneOp.getLidsToRemove();
+    vespalib::IDestructorCallback::SP onDone;
     if (lids_to_remove && lids_to_remove->getNumLids() != 0) {
-        appendOperation(pruneOp, DoneCallback());
-        _activeFeedView->handlePruneRemovedDocuments(pruneOp);
+        appendOperation(pruneOp, onDone);
+        _activeFeedView->handlePruneRemovedDocuments(pruneOp, onDone);
     }
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp b/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp
index 8de727d664f..d2626a0d9f4 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/feedstates.cpp
@@ -20,6 +20,7 @@ using search::transactionlog::client::RPC;
 using search::SerialNum;
 using vespalib::Executor;
 using vespalib::makeLambdaTask;
+using vespalib::IDestructorCallback;
 using vespalib::make_string;
 using proton::bucketdb::IBucketDBHandler;
 
@@ -83,7 +84,7 @@ public:
     }
 
     void replay(const DeleteBucketOperation &op) override {
-        _feed_view_ptr->handleDeleteBucket(op);
+        _feed_view_ptr->handleDeleteBucket(op, IDestructorCallback::SP());
     }
     void replay(const SplitBucketOperation &op) override {
         _bucketDBHandler.handleSplit(op.getSerialNum(), op.getSource(),
@@ -94,15 +95,15 @@ public:
                                     op.getSource2(), op.getTarget());
     }
     void replay(const PruneRemovedDocumentsOperation &op) override {
-        _feed_view_ptr->handlePruneRemovedDocuments(op);
+        _feed_view_ptr->handlePruneRemovedDocuments(op, IDestructorCallback::SP());
     }
     void replay(const MoveOperation &op) override {
-        _feed_view_ptr->handleMove(op, vespalib::IDestructorCallback::SP());
+        _feed_view_ptr->handleMove(op, IDestructorCallback::SP());
     }
     void replay(const CreateBucketOperation &) override {
     }
     void replay(const CompactLidSpaceOperation &op) override {
-        _feed_view_ptr->handleCompactLidSpace(op, vespalib::IDestructorCallback::SP());
+        _feed_view_ptr->handleCompactLidSpace(op, IDestructorCallback::SP());
     }
     NewConfigOperation::IStreamHandler &getNewConfigStreamHandler() override {
         return _config_store;
diff --git a/searchcore/src/vespa/searchcore/proton/server/ifeedview.h b/searchcore/src/vespa/searchcore/proton/server/ifeedview.h
index e0c9fe58c5b..83a91520e5d 100644
--- a/searchcore/src/vespa/searchcore/proton/server/ifeedview.h
+++ b/searchcore/src/vespa/searchcore/proton/server/ifeedview.h
@@ -29,7 +29,8 @@ protected:
     IFeedView() = default;
 public:
     using SP = std::shared_ptr<IFeedView>;
-    using DoneCallback = std::shared_ptr<vespalib::IDestructorCallback>;
+    using DoneCallback = const std::shared_ptr<vespalib::IDestructorCallback> &;
+    using IDestructorCallbackSP = std::shared_ptr<vespalib::IDestructorCallback>;
     using CommitParam = search::CommitParam;
 
     IFeedView(const IFeedView &) = delete;
@@ -55,16 +56,16 @@ public:
     virtual void prepareRemove(RemoveOperation &rmOp) = 0;
     virtual void handleRemove(FeedToken token, const RemoveOperation &rmOp) = 0;
     virtual void prepareDeleteBucket(DeleteBucketOperation &delOp) = 0;
-    virtual void handleDeleteBucket(const DeleteBucketOperation &delOp) = 0;
     virtual void prepareMove(MoveOperation &putOp) = 0;
+    virtual void handleDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone) = 0;
     virtual void handleMove(const MoveOperation &putOp, DoneCallback onDone) = 0;
     virtual void heartBeat(search::SerialNum serialNum, DoneCallback onDone) = 0;
     virtual void forceCommit(const CommitParam & param, DoneCallback onDone) = 0;
-    void forceCommit(CommitParam param) { forceCommit(param, DoneCallback()); }
+    virtual void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation & pruneOp, DoneCallback onDone) = 0;
+    virtual void handleCompactLidSpace(const CompactLidSpaceOperation &op, DoneCallback onDone) = 0;
+    void forceCommit(CommitParam param) { forceCommit(param, IDestructorCallbackSP()); }
     void forceCommit(search::SerialNum serialNum) { forceCommit(CommitParam(serialNum)); }
     void forceCommitAndWait(CommitParam param);
-    virtual void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation & pruneOp) = 0;
-    virtual void handleCompactLidSpace(const CompactLidSpaceOperation &op, DoneCallback onDone) = 0;
 };
 
 } // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.cpp b/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.cpp
index dc7b48f0f02..66b1ba1ae2e 100644
--- a/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.cpp
@@ -157,9 +157,9 @@ SearchableFeedView::removeIndexedFields(SerialNum serialNum, const LidVector &li
 }
 
 void
-SearchableFeedView::internalDeleteBucket(const DeleteBucketOperation &delOp)
+SearchableFeedView::internalDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone)
 {
-    Parent::internalDeleteBucket(delOp);
+    Parent::internalDeleteBucket(delOp, onDone);
 }
 
 void
diff --git a/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.h b/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.h
index 7dd7879a136..70b02c4a39f 100644
--- a/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.h
+++ b/searchcore/src/vespa/searchcore/proton/server/searchable_feed_view.h
@@ -41,7 +41,7 @@ private:
     void performIndexRemove(SerialNum serialNum, const LidVector &lidsToRemove, OnWriteDoneType onWriteDone);
     void performIndexHeartBeat(SerialNum serialNum);
 
-    void internalDeleteBucket(const DeleteBucketOperation &delOp) override;
+    void internalDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone) override;
     void heartBeatIndexedFields(SerialNum serialNum, DoneCallback onDone) override;
 
     void putIndexedFields(SerialNum serialNum, search::DocumentIdT lid, const DocumentSP &newDoc, OnOperationDoneType onWriteDone) override;
diff --git a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp
index 8ca0f232287..97bd940b403 100644
--- a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.cpp
@@ -16,7 +16,6 @@
 #include <vespa/searchcore/proton/reference/i_gid_to_lid_change_handler.h>
 #include <vespa/searchcore/proton/reference/i_pending_gid_to_lid_changes.h>
 #include <vespa/vespalib/util/destructor_callbacks.h>
-#include <vespa/searchlib/common/scheduletaskcallback.h>
 #include <vespa/vespalib/util/isequencedtaskexecutor.h>
 #include <vespa/vespalib/util/exceptions.h>
 
@@ -180,7 +179,7 @@ StoreOnlyFeedView::forceCommit(const CommitParam & param, DoneCallback onDone)
     internalForceCommit(param, std::make_shared<ForceCommitContext>(_writeService.master(), _metaStore,
                                                                     _pendingLidsForCommit->produceSnapshot(),
                                                                     _gidToLidChangeHandler.grab_pending_changes(),
-                                                                    std::move(onDone)));
+                                                                    onDone));
 }
 
 void
@@ -615,7 +614,7 @@ void
 StoreOnlyFeedView::removeIndexedFields(SerialNum , const LidVector &, OnWriteDoneType ) {}
 
 size_t
-StoreOnlyFeedView::removeDocuments(const RemoveDocumentsOperation &op, bool remove_index_and_attributes)
+StoreOnlyFeedView::removeDocuments(const RemoveDocumentsOperation &op, bool remove_index_and_attributes, DoneCallback onWriteDone)
 {
     const SerialNum serialNum = op.getSerialNum();
     const LidVectorContext::SP &ctx = op.getLidsToRemove(_params._subDbId);
@@ -632,10 +631,7 @@ StoreOnlyFeedView::removeDocuments(const RemoveDocumentsOperation &op, bool remo
         _metaStore.removeBatch(lidsToRemove, ctx->getDocIdLimit());
         _lidReuseDelayer.delayReuse(lidsToRemove);
     }
-    std::shared_ptr<vespalib::IDestructorCallback> onWriteDone;
-    vespalib::Executor::Task::UP removeBatchDoneTask;
-    removeBatchDoneTask = makeLambdaTask([]() {});
-    onWriteDone = std::make_shared<search::ScheduleTaskCallback>(_writeService.master(), std::move(removeBatchDoneTask));
+
     if (remove_index_and_attributes) {
         removeIndexedFields(serialNum, lidsToRemove, onWriteDone);
         removeAttributes(serialNum, lidsToRemove, onWriteDone);
@@ -661,15 +657,15 @@ StoreOnlyFeedView::prepareDeleteBucket(DeleteBucketOperation &delOp)
 }
 
 void
-StoreOnlyFeedView::handleDeleteBucket(const DeleteBucketOperation &delOp)
+StoreOnlyFeedView::handleDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone)
 {
-    internalDeleteBucket(delOp);
+    internalDeleteBucket(delOp, onDone);
 }
 
 void
-StoreOnlyFeedView::internalDeleteBucket(const DeleteBucketOperation &delOp)
+StoreOnlyFeedView::internalDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone)
 {
-    size_t rm_count = removeDocuments(delOp, true);
+    size_t rm_count = removeDocuments(delOp, true, onDone);
     LOG(debug, "internalDeleteBucket(): docType(%s), bucket(%s), lidsToRemove(%zu)",
         _params._docTypeName.toString().c_str(), delOp.getBucketId().toString().c_str(), rm_count);
 }
@@ -687,7 +683,7 @@ StoreOnlyFeedView::prepareMove(MoveOperation &moveOp)
 
 // CombiningFeedView calls this for both source and target subdb.
 void
-StoreOnlyFeedView::handleMove(const MoveOperation &moveOp, IDestructorCallback::SP doneCtx)
+StoreOnlyFeedView::handleMove(const MoveOperation &moveOp, DoneCallback doneCtx)
 {
     assert(moveOp.getValidDbdId());
     assert(moveOp.getValidPrevDbdId());
@@ -716,7 +712,7 @@ StoreOnlyFeedView::handleMove(const MoveOperation &moveOp, IDestructorCallback::
         putIndexedFields(serialNum, moveOp.getLid(), doc, onWriteDone);
     }
     if (docAlreadyExists && moveOp.changedDbdId()) {
-        internalRemove(std::move(doneCtx), _pendingLidsForCommit->produce(moveOp.getPrevLid()), serialNum, moveOp.getPrevLid());
+        internalRemove(doneCtx, _pendingLidsForCommit->produce(moveOp.getPrevLid()), serialNum, moveOp.getPrevLid());
     }
 }
 
@@ -734,11 +730,11 @@ StoreOnlyFeedView::heartBeat(SerialNum serialNum, DoneCallback onDone)
 // CombiningFeedView calls this only for the removed subdb.
 void
 StoreOnlyFeedView::
-handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp)
+handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp, DoneCallback onDone)
 {
     assert(_params._subDbType == SubDbType::REMOVED);
     assert(pruneOp.getSubDbId() == _params._subDbId);
-    uint32_t rm_count = removeDocuments(pruneOp, false);
+    uint32_t rm_count = removeDocuments(pruneOp, false, onDone);
 
     LOG(debug, "MinimalFeedView::handlePruneRemovedDocuments called, doctype(%s) %u lids pruned, limit %u",
         _params._docTypeName.toString().c_str(), rm_count,
diff --git a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h
index 4b4082a8aa5..c25accaf4a4 100644
--- a/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h
+++ b/searchcore/src/vespa/searchcore/proton/server/storeonlyfeedview.h
@@ -53,7 +53,7 @@ public:
     using LidVector = LidVectorContext::LidVector;
     using Document = document::Document;
     using DocumentUpdate = document::DocumentUpdate;
-    using OnWriteDoneType = const std::shared_ptr<vespalib::IDestructorCallback> &;
+    using OnWriteDoneType = DoneCallback;
     using OnForceCommitDoneType =const std::shared_ptr<ForceCommitContext> &;
     using OnOperationDoneType = const std::shared_ptr<OperationDoneContext> &;
     using OnPutDoneType = const std::shared_ptr<PutDoneContext> &;
@@ -65,7 +65,6 @@ public:
     using DocumentSP = std::shared_ptr<Document>;
     using DocumentUpdateSP = std::shared_ptr<DocumentUpdate>;
     using LidReuseDelayer = documentmetastore::LidReuseDelayer;
-    using IDestructorCallbackSP = std::shared_ptr<vespalib::IDestructorCallback>;
 
     using Lid = search::DocumentIdT;
 
@@ -179,7 +178,7 @@ private:
 
     // Removes documents from meta store and document store.
     // returns the number of documents removed.
-    size_t removeDocuments(const RemoveDocumentsOperation &op, bool remove_index_and_attribute_fields);
+    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);
 
@@ -189,7 +188,7 @@ private:
                              PromisedDoc promisedDoc, PromisedStream promisedStream);
 
 protected:
-    virtual void internalDeleteBucket(const DeleteBucketOperation &delOp);
+    virtual void internalDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone);
     virtual void heartBeatIndexedFields(SerialNum serialNum, DoneCallback onDone);
     virtual void heartBeatAttributes(SerialNum serialNum, DoneCallback onDone);
 
@@ -231,9 +230,9 @@ public:
     void prepareRemove(RemoveOperation &rmOp) override;
     void handleRemove(FeedToken token, const RemoveOperation &rmOp) override;
     void prepareDeleteBucket(DeleteBucketOperation &delOp) override;
-    void handleDeleteBucket(const DeleteBucketOperation &delOp) override;
+    void handleDeleteBucket(const DeleteBucketOperation &delOp, DoneCallback onDone) override;
     void prepareMove(MoveOperation &putOp) override;
-    void handleMove(const MoveOperation &putOp, std::shared_ptr<vespalib::IDestructorCallback> doneCtx) override;
+    void handleMove(const MoveOperation &putOp, DoneCallback doneCtx) override;
     void heartBeat(search::SerialNum serialNum, DoneCallback onDone) override;
     void forceCommit(const CommitParam & param, DoneCallback onDone) override;
 
@@ -243,7 +242,7 @@ public:
      *
      * Called by writer thread.
      */
-    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp) override;
+    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &pruneOp, DoneCallback onDone) override;
     void handleCompactLidSpace(const CompactLidSpaceOperation &op, DoneCallback onDone) override;
     std::shared_ptr<PendingLidTrackerBase> getUncommittedLidTracker() { return _pendingLidsForCommit; }
 };
diff --git a/searchcore/src/vespa/searchcore/proton/test/dummy_feed_view.h b/searchcore/src/vespa/searchcore/proton/test/dummy_feed_view.h
index 7f11bf9ec51..51bb3ebc807 100644
--- a/searchcore/src/vespa/searchcore/proton/test/dummy_feed_view.h
+++ b/searchcore/src/vespa/searchcore/proton/test/dummy_feed_view.h
@@ -25,11 +25,11 @@ struct DummyFeedView : public IFeedView
     void prepareRemove(RemoveOperation &) override {}
     void handleRemove(FeedToken, const RemoveOperation &) override {}
     void prepareDeleteBucket(DeleteBucketOperation &) override {}
-    void handleDeleteBucket(const DeleteBucketOperation &) override {}
+    void handleDeleteBucket(const DeleteBucketOperation &, DoneCallback) override {}
     void prepareMove(MoveOperation &) override {}
-    void handleMove(const MoveOperation &, std::shared_ptr<vespalib::IDestructorCallback>) override {}
+    void handleMove(const MoveOperation &, DoneCallback) override {}
     void heartBeat(search::SerialNum, DoneCallback) override {}
-    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &) override {}
+    void handlePruneRemovedDocuments(const PruneRemovedDocumentsOperation &, DoneCallback) override {}
     void handleCompactLidSpace(const CompactLidSpaceOperation &, DoneCallback) override {}
     void forceCommit(const CommitParam &, DoneCallback) override { }
 };