Add backend support for distributed condition evaluation

Lets the "test" part of a test-and-set condition be evaluated
locally on individual content nodes. Piggybacks on top of metadata-only
Get operations, adding a new condition field to the request and a
boolean match result to the response.

Decouples the existing TaS utility code from being command-oriented,
allowing it to be used in other contexts as well.

Not yet wired through any protocols.

11 files changed, 245 insertions, 61 deletions

diff --git a/storage/src/tests/persistence/persistencetestutils.h b/storage/src/tests/persistence/persistencetestutils.h
index 94ae7b9fb53..e60260f3ee8 100644
--- a/storage/src/tests/persistence/persistencetestutils.h
+++ b/storage/src/tests/persistence/persistencetestutils.h
@@ -150,6 +150,18 @@ public:
                                                 _replySender, MockBucketLock::make(bucket, _mock_bucket_locks), std::move(cmd));
     }
 
+    template <typename T>
+    requires std::is_base_of_v<api::StorageReply, T>
+    [[nodiscard]] std::shared_ptr<T>
+    fetch_single_reply(MessageTracker::UP tracker) {
+        if (tracker && tracker->hasReply()) {
+            tracker->sendReply(); // Forward to queue so we can fetch it below
+        }
+        std::shared_ptr<api::StorageMessage> msg;
+        _replySender.queue.getNext(msg, 60s);
+        return std::dynamic_pointer_cast<T>(msg);
+    }
+
     api::ReturnCode
     fetchResult(const MessageTracker::UP & tracker) {
         if (tracker) {
diff --git a/storage/src/tests/persistence/testandsettest.cpp b/storage/src/tests/persistence/testandsettest.cpp
index 5be1c7cd92a..1aa359de634 100644
--- a/storage/src/tests/persistence/testandsettest.cpp
+++ b/storage/src/tests/persistence/testandsettest.cpp
@@ -1,16 +1,16 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 // @author Vegard Sjonfjell
-#include <vespa/storage/persistence/persistencehandler.h>
 #include <tests/persistence/persistencetestutils.h>
 #include <vespa/document/test/make_document_bucket.h>
-#include <vespa/documentapi/messagebus/messages/testandsetcondition.h>
 #include <vespa/document/fieldvalue/fieldvalues.h>
 #include <vespa/document/update/documentupdate.h>
 #include <vespa/document/update/assignvalueupdate.h>
 #include <vespa/document/fieldset/fieldsets.h>
+#include <vespa/documentapi/messagebus/messages/testandsetcondition.h>
 #include <vespa/persistence/spi/test.h>
 #include <vespa/persistence/spi/persistenceprovider.h>
 #include <vespa/persistence/spi/docentry.h>
+#include <vespa/storage/persistence/persistencehandler.h>
 #include <functional>
 
 using std::unique_ptr;
@@ -19,6 +19,7 @@ using std::shared_ptr;
 using storage::spi::test::makeSpiBucket;
 using document::test::makeDocumentBucket;
 using document::StringFieldValue;
+using documentapi::TestAndSetCondition;
 using namespace ::testing;
 
 namespace storage {
@@ -34,15 +35,18 @@ struct TestAndSetTest : PersistenceTestUtils {
     const StringFieldValue OLD_CONTENT{"Some old content"};
     const StringFieldValue NEW_CONTENT{"Freshly pressed and squeezed content"};
     const document::Bucket BUCKET = makeDocumentBucket(BUCKET_ID);
+    const TestAndSetCondition MATCHING_CONDITION{"testdoctype1.hstringval=\"*woofy dog*\""};
 
     unique_ptr<PersistenceHandler> persistenceHandler;
     const AsyncHandler * asyncHandler;
+    const SimpleMessageHandler* simple_handler;
     shared_ptr<document::Document> testDoc;
     document::DocumentId testDocId;
 
     TestAndSetTest()
         : persistenceHandler(),
-          asyncHandler(nullptr)
+          asyncHandler(nullptr),
+          simple_handler(nullptr)
     {}
 
     void SetUp() override {
@@ -54,6 +58,7 @@ struct TestAndSetTest : PersistenceTestUtils {
         testDoc = createTestDocument();
         testDocId = testDoc->getId();
         asyncHandler = &_persistenceHandler->asyncHandler();
+        simple_handler = &_persistenceHandler->simpleMessageHandler();
     }
 
     void TearDown() override {
@@ -68,6 +73,8 @@ struct TestAndSetTest : PersistenceTestUtils {
     document::Document::SP retrieveTestDocument();
     void setTestCondition(api::TestAndSetCommand & command);
     void putTestDocument(bool matchingHeader, api::Timestamp timestamp);
+    std::shared_ptr<api::GetReply> invoke_conditional_get();
+    void feed_remove_entry_with_timestamp(api::Timestamp timestamp);
     void assertTestDocumentFoundAndMatchesContent(const document::FieldValue & value);
 
     static std::string expectedDocEntryString(
@@ -247,6 +254,59 @@ TEST_F(TestAndSetTest, conditional_put_to_non_existing_document_should_fail) {
     EXPECT_EQ("", dumpBucket(BUCKET_ID));
 }
 
+TEST_F(TestAndSetTest, conditional_get_returns_doc_metadata_on_match) {
+    const api::Timestamp timestamp = 12345;
+    putTestDocument(true, timestamp);
+    auto reply = invoke_conditional_get();
+
+    ASSERT_EQ(reply->getResult(), api::ReturnCode());
+    EXPECT_EQ(reply->getLastModifiedTimestamp(), timestamp);
+    EXPECT_TRUE(reply->condition_matched());
+    EXPECT_FALSE(reply->is_tombstone());
+    // Checking reply->wasFound() is tempting but doesn't make sense here, as that checks for
+    // the presence of a document object, which metadata-only gets by definition do not return.
+}
+
+TEST_F(TestAndSetTest, conditional_get_returns_doc_metadata_on_mismatch) {
+    const api::Timestamp timestamp = 12345;
+    putTestDocument(false, timestamp);
+    auto reply = invoke_conditional_get();
+
+    ASSERT_EQ(reply->getResult(), api::ReturnCode());
+    EXPECT_EQ(reply->getLastModifiedTimestamp(), timestamp);
+    EXPECT_FALSE(reply->condition_matched());
+    EXPECT_FALSE(reply->is_tombstone());
+}
+
+TEST_F(TestAndSetTest, conditional_get_for_non_existing_document_returns_zero_timestamp) {
+    auto reply = invoke_conditional_get();
+
+    ASSERT_EQ(reply->getResult(), api::ReturnCode());
+    EXPECT_EQ(reply->getLastModifiedTimestamp(), 0);
+    EXPECT_FALSE(reply->condition_matched());
+    EXPECT_FALSE(reply->is_tombstone());
+}
+
+TEST_F(TestAndSetTest, conditional_get_for_non_existing_document_with_explicit_tombstone_returns_tombstone_timestamp) {
+    api::Timestamp timestamp = 56789;
+    feed_remove_entry_with_timestamp(timestamp);
+    auto reply = invoke_conditional_get();
+
+    ASSERT_EQ(reply->getResult(), api::ReturnCode());
+    EXPECT_EQ(reply->getLastModifiedTimestamp(), timestamp);
+    EXPECT_FALSE(reply->condition_matched());
+    EXPECT_TRUE(reply->is_tombstone());
+}
+
+TEST_F(TestAndSetTest, conditional_get_requires_metadata_only_fieldset) {
+    auto get = std::make_shared<api::GetCommand>(BUCKET, testDocId, document::AllFields::NAME);
+    get->set_condition(MATCHING_CONDITION);
+    // Note: uses fetchResult instead of fetch_single_reply due to implicit failure signalling via tracker instance.
+    auto result = fetchResult(simple_handler->handleGet(*get, createTracker(get, BUCKET)));
+    ASSERT_EQ(result, api::ReturnCode(api::ReturnCode::ILLEGAL_PARAMETERS,
+                                      "Conditional Get operations must be metadata-only"));
+}
+
 document::Document::SP
 TestAndSetTest::createTestDocument()
 {
@@ -270,7 +330,7 @@ TestAndSetTest::retrieveTestDocument()
     auto tracker = _persistenceHandler->simpleMessageHandler().handleGet(*get, createTracker(get, BUCKET));
     assert(tracker->getResult() == api::ReturnCode::Result::OK);
 
-    auto & reply = static_cast<api::GetReply &>(tracker->getReply());
+    auto& reply = dynamic_cast<api::GetReply&>(tracker->getReply());
     assert(reply.wasFound());
 
     return reply.getDocument();
@@ -278,7 +338,7 @@ TestAndSetTest::retrieveTestDocument()
 
 void TestAndSetTest::setTestCondition(api::TestAndSetCommand & command)
 {
-    command.setCondition(documentapi::TestAndSetCondition("testdoctype1.hstringval=\"*woofy dog*\""));
+    command.setCondition(MATCHING_CONDITION);
 }
 
 void TestAndSetTest::putTestDocument(bool matchingHeader, api::Timestamp timestamp) {
@@ -290,6 +350,17 @@ void TestAndSetTest::putTestDocument(bool matchingHeader, api::Timestamp timesta
     fetchResult(asyncHandler->handlePut(*put, createTracker(put, BUCKET)));
 }
 
+std::shared_ptr<api::GetReply> TestAndSetTest::invoke_conditional_get() {
+    auto get = std::make_shared<api::GetCommand>(BUCKET, testDocId, document::NoFields::NAME);
+    get->set_condition(MATCHING_CONDITION);
+    return fetch_single_reply<api::GetReply>(simple_handler->handleGet(*get, createTracker(get, BUCKET)));
+}
+
+void TestAndSetTest::feed_remove_entry_with_timestamp(api::Timestamp timestamp) {
+    auto remove = std::make_shared<api::RemoveCommand>(BUCKET, testDocId, timestamp);
+    (void)fetchResult(asyncHandler->handleRemove(*remove, createTracker(remove, BUCKET)));
+}
+
 void TestAndSetTest::assertTestDocumentFoundAndMatchesContent(const document::FieldValue & value)
 {
     auto doc = retrieveTestDocument();
diff --git a/storage/src/tests/storageapi/mbusprot/storageprotocoltest.cpp b/storage/src/tests/storageapi/mbusprot/storageprotocoltest.cpp
index d3036a2fad3..6d8c3585726 100644
--- a/storage/src/tests/storageapi/mbusprot/storageprotocoltest.cpp
+++ b/storage/src/tests/storageapi/mbusprot/storageprotocoltest.cpp
@@ -848,7 +848,7 @@ TEST_P(StorageProtocolTest, track_memory_footprint_for_some_messages) {
     EXPECT_EQ(144u + sizeof(vespalib::string), sizeof(PutCommand));
     EXPECT_EQ(144u + sizeof(vespalib::string), sizeof(UpdateCommand));
     EXPECT_EQ(224u + sizeof(vespalib::string), sizeof(RemoveCommand));
-    EXPECT_EQ(296u, sizeof(GetCommand));
+    EXPECT_EQ(296u + sizeof(documentapi::TestAndSetCondition), sizeof(GetCommand));
 }
 
 } // storage::api
diff --git a/storage/src/vespa/storage/persistence/asynchandler.cpp b/storage/src/vespa/storage/persistence/asynchandler.cpp
index e20c0475556..60c6d507416 100644
--- a/storage/src/vespa/storage/persistence/asynchandler.cpp
+++ b/storage/src/vespa/storage/persistence/asynchandler.cpp
@@ -358,7 +358,11 @@ bool
 AsyncHandler::tasConditionMatches(const api::TestAndSetCommand & cmd, MessageTracker & tracker,
                                   spi::Context & context, bool missingDocumentImpliesMatch) const {
     try {
-        TestAndSetHelper helper(_env, _spi, _bucketIdFactory, cmd, missingDocumentImpliesMatch);
+        TestAndSetHelper helper(_env, _spi, _bucketIdFactory,
+                                cmd.getCondition(),
+                                cmd.getBucket(), cmd.getDocumentId(),
+                                cmd.getDocumentType(),
+                                missingDocumentImpliesMatch);
 
         auto code = helper.retrieveAndMatch(context);
         if (code.failed()) {
diff --git a/storage/src/vespa/storage/persistence/persistencehandler.cpp b/storage/src/vespa/storage/persistence/persistencehandler.cpp
index 8d71cc9308b..69f910d0910 100644
--- a/storage/src/vespa/storage/persistence/persistencehandler.cpp
+++ b/storage/src/vespa/storage/persistence/persistencehandler.cpp
@@ -24,7 +24,7 @@ PersistenceHandler::PersistenceHandler(vespalib::ISequencedTaskExecutor & sequen
                     cfg.commonMergeChainOptimalizationMinimumSize),
       _asyncHandler(_env, provider, bucketOwnershipNotifier, sequencedExecutor, component.getBucketIdFactory()),
       _splitJoinHandler(_env, provider, bucketOwnershipNotifier, cfg.enableMultibitSplitOptimalization),
-      _simpleHandler(_env, provider)
+      _simpleHandler(_env, provider, component.getBucketIdFactory())
 {
 }
 
diff --git a/storage/src/vespa/storage/persistence/simplemessagehandler.cpp b/storage/src/vespa/storage/persistence/simplemessagehandler.cpp
index e83d460f47a..ea929bf8620 100644
--- a/storage/src/vespa/storage/persistence/simplemessagehandler.cpp
+++ b/storage/src/vespa/storage/persistence/simplemessagehandler.cpp
@@ -2,6 +2,7 @@
 
 #include "simplemessagehandler.h"
 #include "persistenceutil.h"
+#include "testandsethelper.h"
 #include <vespa/persistence/spi/persistenceprovider.h>
 #include <vespa/persistence/spi/docentry.h>
 #include <vespa/storageapi/message/bucket.h>
@@ -45,21 +46,45 @@ getFieldSet(const document::FieldSetRepo & repo, vespalib::stringref name, Messa
 }
 
 }
-SimpleMessageHandler::SimpleMessageHandler(const PersistenceUtil& env, spi::PersistenceProvider& spi)
+SimpleMessageHandler::SimpleMessageHandler(const PersistenceUtil& env,
+                                           spi::PersistenceProvider& spi,
+                                           const document::BucketIdFactory& bucket_id_factory)
     : _env(env),
-      _spi(spi)
+      _spi(spi),
+      _bucket_id_factory(bucket_id_factory)
 {
 }
 
 MessageTracker::UP
+SimpleMessageHandler::handle_conditional_get(api::GetCommand& cmd, MessageTracker::UP tracker) const
+{
+    if (cmd.getFieldSet() == document::NoFields::NAME) {
+        TestAndSetHelper tas_helper(_env, _spi, _bucket_id_factory, cmd.condition(),
+                                    cmd.getBucket(), cmd.getDocumentId(), nullptr);
+        auto result = tas_helper.fetch_and_match_raw(tracker->context());
+        tracker->setReply(std::make_shared<api::GetReply>(cmd, nullptr, result.timestamp, false,
+                                                          result.is_tombstone(), result.is_match()));
+    } else {
+        tracker->fail(api::ReturnCode::ILLEGAL_PARAMETERS, "Conditional Get operations must be metadata-only");
+    }
+    return tracker;
+}
+
+MessageTracker::UP
 SimpleMessageHandler::handleGet(api::GetCommand& cmd, MessageTracker::UP tracker) const
 {
     auto& metrics = _env._metrics.get;
     tracker->setMetric(metrics);
     metrics.request_size.addValue(cmd.getApproxByteSize());
 
+    if (cmd.has_condition()) {
+        return handle_conditional_get(cmd, std::move(tracker));
+    }
+
     auto fieldSet = getFieldSet(_env.getFieldSetRepo(), cmd.getFieldSet(), *tracker);
-    if ( ! fieldSet) { return tracker; }
+    if (!fieldSet) {
+        return tracker;
+    }
 
     tracker->context().setReadConsistency(api_read_consistency_to_spi(cmd.internal_read_consistency()));
     spi::GetResult result = _spi.get(_env.getBucket(cmd.getDocumentId(), cmd.getBucket()),
@@ -70,7 +95,7 @@ SimpleMessageHandler::handleGet(api::GetCommand& cmd, MessageTracker::UP tracker
             metrics.notFound.inc();
         }
         tracker->setReply(std::make_shared<api::GetReply>(cmd, result.getDocumentPtr(), result.getTimestamp(),
-                                                          false, result.is_tombstone()));
+                                                          false, result.is_tombstone(), false));
     }
 
     return tracker;
diff --git a/storage/src/vespa/storage/persistence/simplemessagehandler.h b/storage/src/vespa/storage/persistence/simplemessagehandler.h
index 009fd6dff52..a5a19772556 100644
--- a/storage/src/vespa/storage/persistence/simplemessagehandler.h
+++ b/storage/src/vespa/storage/persistence/simplemessagehandler.h
@@ -7,6 +7,8 @@
 #include <vespa/storage/common/bucketmessages.h>
 #include <vespa/storageapi/message/persistence.h>
 
+namespace document { class BucketIdFactory; }
+
 namespace storage {
 
 namespace spi { struct PersistenceProvider; }
@@ -19,7 +21,9 @@ class PersistenceUtil;
  */
 class SimpleMessageHandler : public Types {
 public:
-    SimpleMessageHandler(const PersistenceUtil&, spi::PersistenceProvider&);
+    SimpleMessageHandler(const PersistenceUtil&,
+                         spi::PersistenceProvider&,
+                         const document::BucketIdFactory&);
     MessageTrackerUP handleGet(api::GetCommand& cmd, MessageTrackerUP tracker) const;
     MessageTrackerUP handleRevert(api::RevertCommand& cmd, MessageTrackerUP tracker) const;
     MessageTrackerUP handleCreateIterator(CreateIteratorCommand& cmd, MessageTrackerUP tracker) const;
@@ -27,8 +31,11 @@ public:
     MessageTrackerUP handleReadBucketList(ReadBucketList& cmd, MessageTrackerUP tracker) const;
     MessageTrackerUP handleReadBucketInfo(ReadBucketInfo& cmd, MessageTrackerUP tracker) const;
 private:
-    const PersistenceUtil    & _env;
-    spi::PersistenceProvider & _spi;
+    MessageTrackerUP handle_conditional_get(api::GetCommand& cmd, MessageTrackerUP tracker) const;
+
+    const PersistenceUtil&           _env;
+    spi::PersistenceProvider&        _spi;
+    const document::BucketIdFactory& _bucket_id_factory;
 };
 
 } // storage
diff --git a/storage/src/vespa/storage/persistence/testandsethelper.cpp b/storage/src/vespa/storage/persistence/testandsethelper.cpp
index 393dac09f72..1cda9427761 100644
--- a/storage/src/vespa/storage/persistence/testandsethelper.cpp
+++ b/storage/src/vespa/storage/persistence/testandsethelper.cpp
@@ -31,69 +31,91 @@ void TestAndSetHelper::parseDocumentSelection(const document::DocumentTypeRepo &
     document::select::Parser parser(documentTypeRepo, bucketIdFactory);
 
     try {
-        _docSelectionUp = parser.parse(_cmd.getCondition().getSelection());
+        _docSelectionUp = parser.parse(_condition.getSelection());
     } catch (const document::select::ParsingFailedException & e) {
         throw TestAndSetException(api::ReturnCode(api::ReturnCode::ILLEGAL_PARAMETERS, "Failed to parse test and set condition: "s + e.getMessage()));
     }
 }
 
 spi::GetResult TestAndSetHelper::retrieveDocument(const document::FieldSet & fieldSet, spi::Context & context) {
-    return _spi.get(_env.getBucket(_docId, _cmd.getBucket()), fieldSet, _cmd.getDocumentId(), context);
+    return _spi.get(_env.getBucket(_docId, _bucket), fieldSet, _docId, context);
 }
 
-TestAndSetHelper::TestAndSetHelper(const PersistenceUtil & env, const spi::PersistenceProvider  & spi,
-                                   const document::BucketIdFactory & bucketFactory,
-                                   const api::TestAndSetCommand & cmd, bool missingDocumentImpliesMatch)
+TestAndSetHelper::TestAndSetHelper(const PersistenceUtil& env,
+                                   const spi::PersistenceProvider& spi,
+                                   const document::BucketIdFactory& bucket_id_factory,
+                                   const documentapi::TestAndSetCondition& condition,
+                                   document::Bucket bucket,
+                                   document::DocumentId doc_id,
+                                   const document::DocumentType* doc_type_ptr,
+                                   bool missingDocumentImpliesMatch)
     : _env(env),
       _spi(spi),
-      _cmd(cmd),
-      _docId(cmd.getDocumentId()),
-      _docTypePtr(_cmd.getDocumentType()),
+      _condition(condition),
+      _bucket(bucket),
+      _docId(std::move(doc_id)),
+      _docTypePtr(doc_type_ptr),
       _missingDocumentImpliesMatch(missingDocumentImpliesMatch)
 {
     const auto & repo = _env.getDocumentTypeRepo();
     resolveDocumentType(repo);
-    parseDocumentSelection(repo, bucketFactory);
+    parseDocumentSelection(repo, bucket_id_factory);
 }
 
 TestAndSetHelper::~TestAndSetHelper() = default;
 
-api::ReturnCode
-TestAndSetHelper::retrieveAndMatch(spi::Context & context) {
-    // Walk document selection tree to build a minimal field set 
+TestAndSetHelper::Result
+TestAndSetHelper::fetch_and_match_raw(spi::Context& context) {
+    // Walk document selection tree to build a minimal field set
     FieldVisitor fieldVisitor(*_docTypePtr);
     try {
         _docSelectionUp->visit(fieldVisitor);
     } catch (const document::FieldNotFoundException& e) {
-        return api::ReturnCode(api::ReturnCode::ILLEGAL_PARAMETERS,
-                               vespalib::make_string("Condition field '%s' could not be found, or is an imported field. "
-                                                     "Imported fields are not supported in conditional mutations.",
-                                                     e.getFieldName().c_str()));
+        throw TestAndSetException(api::ReturnCode(
+                api::ReturnCode::ILLEGAL_PARAMETERS,
+                vespalib::make_string("Condition field '%s' could not be found, or is an imported field. "
+                                      "Imported fields are not supported in conditional mutations.",
+                                      e.getFieldName().c_str())));
     }
-
-    // Retrieve document
     auto result = retrieveDocument(fieldVisitor.getFieldSet(), context);
-
     // If document exists, match it with selection
     if (result.hasDocument()) {
         auto docPtr = result.getDocumentPtr();
         if (_docSelectionUp->contains(*docPtr) != document::select::Result::True) {
-            return api::ReturnCode(api::ReturnCode::TEST_AND_SET_CONDITION_FAILED,
-                                   vespalib::make_string("Condition did not match document nodeIndex=%d bucket=%" PRIx64 " %s",
-                                                         _env._nodeIndex, _cmd.getBucketId().getRawId(),
-                                                         _cmd.hasBeenRemapped() ? "remapped" : ""));
+            return {result.getTimestamp(), Result::ConditionOutcome::IsNotMatch};
         }
-
         // Document matches
-        return api::ReturnCode();
-    } else if (_missingDocumentImpliesMatch) {
-        return api::ReturnCode();
+        return {result.getTimestamp(), Result::ConditionOutcome::IsMatch};
     }
+    return {result.getTimestamp(), result.is_tombstone() ? Result::ConditionOutcome::IsTombstone
+                                                         : Result::ConditionOutcome::DocNotFound};
+}
 
-    return api::ReturnCode(api::ReturnCode::TEST_AND_SET_CONDITION_FAILED,
-                           vespalib::make_string("Document does not exist nodeIndex=%d bucket=%" PRIx64 " %s",
-                                                 _env._nodeIndex, _cmd.getBucketId().getRawId(),
-                                                 _cmd.hasBeenRemapped() ? "remapped" : ""));
+api::ReturnCode
+TestAndSetHelper::to_api_return_code(const Result& result) const {
+    switch (result.condition_outcome) {
+    case Result::ConditionOutcome::IsNotMatch:
+        return {api::ReturnCode::TEST_AND_SET_CONDITION_FAILED,
+                vespalib::make_string("Condition did not match document nodeIndex=%d bucket=%" PRIx64,
+                                      _env._nodeIndex, _bucket.getBucketId().getRawId())};
+    case Result::ConditionOutcome::IsTombstone:
+    case Result::ConditionOutcome::DocNotFound:
+        if (!_missingDocumentImpliesMatch) {
+            return {api::ReturnCode::TEST_AND_SET_CONDITION_FAILED,
+                    vespalib::make_string("Document does not exist nodeIndex=%d bucket=%" PRIx64,
+                                          _env._nodeIndex, _bucket.getBucketId().getRawId())};
+        }
+        [[fallthrough]]; // as match
+    case Result::ConditionOutcome::IsMatch:
+        return {}; // OK
+    }
+    abort();
+}
+
+api::ReturnCode
+TestAndSetHelper::retrieveAndMatch(spi::Context & context) {
+    auto result = fetch_and_match_raw(context);
+    return to_api_return_code(result);
 }
 
 } // storage
diff --git a/storage/src/vespa/storage/persistence/testandsethelper.h b/storage/src/vespa/storage/persistence/testandsethelper.h
index 82710e523c4..31b1cc79a54 100644
--- a/storage/src/vespa/storage/persistence/testandsethelper.h
+++ b/storage/src/vespa/storage/persistence/testandsethelper.h
@@ -25,9 +25,8 @@ class PersistenceUtil;
 
 class TestAndSetException : public std::runtime_error {
     api::ReturnCode _code;
-
 public:
-    TestAndSetException(api::ReturnCode code)
+    explicit TestAndSetException(api::ReturnCode code)
         : std::runtime_error(code.getMessage()),
         _code(std::move(code))
     {}
@@ -36,11 +35,12 @@ public:
 };
 
 class TestAndSetHelper {
-    const PersistenceUtil                  &_env;
-    const spi::PersistenceProvider         &_spi;
-    const api::TestAndSetCommand           &_cmd;
+    const PersistenceUtil&                  _env;
+    const spi::PersistenceProvider&         _spi;
+    const documentapi::TestAndSetCondition& _condition;
+    const document::Bucket                  _bucket;
     const document::DocumentId              _docId;
-    const document::DocumentType *          _docTypePtr;
+    const document::DocumentType*           _docTypePtr;
     std::unique_ptr<document::select::Node> _docSelectionUp;
     bool                                    _missingDocumentImpliesMatch;
 
@@ -50,10 +50,44 @@ class TestAndSetHelper {
     spi::GetResult retrieveDocument(const document::FieldSet & fieldSet, spi::Context & context);
 
 public:
-    TestAndSetHelper(const PersistenceUtil & env, const spi::PersistenceProvider & _spi,
-                     const document::BucketIdFactory & bucketIdFactory,
-                     const api::TestAndSetCommand & cmd, bool missingDocumentImpliesMatch = false);
+    struct Result {
+        enum class ConditionOutcome {
+            DocNotFound,
+            IsMatch,
+            IsNotMatch,
+            IsTombstone
+        };
+
+        api::Timestamp timestamp = 0;
+        ConditionOutcome condition_outcome = ConditionOutcome::IsNotMatch;
+
+        [[nodiscard]] bool doc_not_found() const noexcept {
+            return condition_outcome == ConditionOutcome::DocNotFound;
+        }
+        [[nodiscard]] bool is_match() const noexcept {
+            return condition_outcome == ConditionOutcome::IsMatch;
+        }
+        [[nodiscard]] bool is_not_match() const noexcept {
+            return condition_outcome == ConditionOutcome::IsNotMatch;
+        }
+        [[nodiscard]] bool is_tombstone() const noexcept {
+            return condition_outcome == ConditionOutcome::IsTombstone;
+        }
+    };
+
+    TestAndSetHelper(const PersistenceUtil& env,
+                     const spi::PersistenceProvider& _spi,
+                     const document::BucketIdFactory& bucket_id_factory,
+                     const documentapi::TestAndSetCondition& condition,
+                     document::Bucket bucket,
+                     document::DocumentId doc_id,
+                     const document::DocumentType* doc_type_ptr,
+                     bool missingDocumentImpliesMatch = false);
     ~TestAndSetHelper();
+
+    Result fetch_and_match_raw(spi::Context& context);
+    api::ReturnCode to_api_return_code(const Result& result) const;
+
     api::ReturnCode retrieveAndMatch(spi::Context & context);
 };
 
diff --git a/storage/src/vespa/storageapi/message/persistence.cpp b/storage/src/vespa/storageapi/message/persistence.cpp
index 41a53449b67..1b09639fd9b 100644
--- a/storage/src/vespa/storageapi/message/persistence.cpp
+++ b/storage/src/vespa/storageapi/message/persistence.cpp
@@ -222,7 +222,8 @@ GetReply::GetReply(const GetCommand& cmd,
                    const DocumentSP& doc,
                    Timestamp lastModified,
                    bool had_consistent_replicas,
-                   bool is_tombstone)
+                   bool is_tombstone,
+                   bool condition_matched)
     : BucketInfoReply(cmd),
       _docId(cmd.getDocumentId()),
       _fieldSet(cmd.getFieldSet()),
@@ -230,7 +231,8 @@ GetReply::GetReply(const GetCommand& cmd,
       _beforeTimestamp(cmd.getBeforeTimestamp()),
       _lastModifiedTime(lastModified),
       _had_consistent_replicas(had_consistent_replicas),
-      _is_tombstone(is_tombstone)
+      _is_tombstone(is_tombstone),
+      _condition_matched(condition_matched)
 {
 }
 
diff --git a/storage/src/vespa/storageapi/message/persistence.h b/storage/src/vespa/storageapi/message/persistence.h
index d1709c46a6e..d010c295ca7 100644
--- a/storage/src/vespa/storageapi/message/persistence.h
+++ b/storage/src/vespa/storageapi/message/persistence.h
@@ -185,9 +185,10 @@ public:
  * timestamp.
  */
 class GetCommand : public BucketInfoCommand {
-    document::DocumentId _docId;
-    Timestamp _beforeTimestamp;
-    vespalib::string _fieldSet;
+    document::DocumentId    _docId;
+    Timestamp               _beforeTimestamp;
+    vespalib::string        _fieldSet;
+    TestAndSetCondition     _condition;
     InternalReadConsistency _internal_read_consistency;
 public:
     GetCommand(const document::Bucket &bucket, const document::DocumentId&,
@@ -198,6 +199,9 @@ public:
     Timestamp getBeforeTimestamp() const { return _beforeTimestamp; }
     const vespalib::string& getFieldSet() const { return _fieldSet; }
     void setFieldSet(vespalib::stringref fieldSet) { _fieldSet = fieldSet; }
+    [[nodiscard]] bool has_condition() const noexcept { return _condition.isPresent(); }
+    [[nodiscard]] const TestAndSetCondition& condition() const noexcept { return _condition; }
+    void set_condition(TestAndSetCondition cond) { _condition = std::move(cond); }
     InternalReadConsistency internal_read_consistency() const noexcept {
         return _internal_read_consistency;
     }
@@ -229,12 +233,14 @@ class GetReply : public BucketInfoReply {
     Timestamp _lastModifiedTime;
     bool _had_consistent_replicas;
     bool _is_tombstone;
+    bool _condition_matched;
 public:
     explicit GetReply(const GetCommand& cmd,
                       const DocumentSP& doc = DocumentSP(),
                       Timestamp lastModified = 0,
                       bool had_consistent_replicas = false,
-                      bool is_tombstone = false);
+                      bool is_tombstone = false,
+                      bool condition_matched = false);
 
     ~GetReply() override;
 
@@ -247,6 +253,7 @@ public:
 
     [[nodiscard]] bool had_consistent_replicas() const noexcept { return _had_consistent_replicas; }
     [[nodiscard]] bool is_tombstone() const noexcept { return _is_tombstone; }
+    [[nodiscard]] bool condition_matched() const noexcept { return _condition_matched; }
 
     bool wasFound() const { return (_doc.get() != nullptr); }
     void print(std::ostream& out, bool verbose, const std::string& indent) const override;