Log the first failed CAS blob publication attempt below Warning

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp

@@ -78,14 +78,17 @@ uint64_t nowMs()
 }
 
 /// Rate-limited, because under an outage every blob of every write fails here; the counter has them all.
+/// The first failure is routine and the next attempt usually succeeds, so only a repeat is a warning.
 void recordFailedPublicationAttempt(
     const String & key, BlobPublicationTransport transport, int attempt, int max_attempts, const String & error)
 {
     ProfileEvents::increment(ProfileEvents::CASBlobPublishAttemptFailed);
     LogSeriesLimiter log(getLogger("CasBlobPublishAttempt"), /*allowed_count=*/5, /*interval_s=*/30);
-    LOG_WARNING(log, "Publication of CAS blob {} ({}) failed, attempt {} of {}: {}",
-                key, transport == BlobPublicationTransport::Streaming ? "streaming" : "server-side copy",
-                attempt + 1, max_attempts, error);
+    const char * transport_name = transport == BlobPublicationTransport::Streaming ? "streaming" : "server-side copy";
+    if (attempt == 0)
+        LOG_INFO(log, "Publication of CAS blob {} ({}) failed, attempt {} of {}: {}", key, transport_name, attempt + 1, max_attempts, error);
+    else
+        LOG_WARNING(log, "Publication of CAS blob {} ({}) failed, attempt {} of {}: {}", key, transport_name, attempt + 1, max_attempts, error);
 }
 
 bool isDeterministicBlobPublicationFailure(const std::exception & error)

src/Disks/tests/gtest_cas_blob_publish_transport.cpp

@@ -188,6 +188,16 @@ CountingS3::ObjectStore::FaultFn dropFirst(Verb verb)
     };
 }
 
+CountingS3::ObjectStore::FaultFn dropFirstTwo(Verb verb)
+{
+    return [verb](const CountingS3::Request & request, Poco::Net::HTTPServerResponse &)
+    {
+        if (request.verb == verb && request.nth <= 2)
+            throw CountingS3::DropResponse();
+        return false;
+    };
+}
+
 uint64_t failedPublicationAttempts()
 {
     return ProfileEvents::global_counters[ProfileEvents::CASBlobPublishAttemptFailed].load();
@@ -224,6 +234,27 @@ TEST(CASBlobPublishTransport, LostPutIsRepublishedWithAFreshEnvelope)
     EXPECT_EQ(failedPublicationAttempts() - failed_before, 1u);
 }
 
+/// A single lost attempt is routine and the next one usually succeeds, so only a repeated failure is
+/// worth a client-visible warning.
+TEST(CASBlobPublishTransport, FirstFailedAttemptIsBelowWarning)
+{
+    (void)getContext();
+    CountingS3::ObjectStore store(dropFirstTwo(Verb::PutObject));
+    CountingS3Server server(store.responder());
+    auto pool = openPool(server.getUrl());
+    const std::string payload = "two-lost-puts-payload";
+    auto build = startBuild(pool, payload);
+    DB::Cas::tests::ScopedLogCapture capture("CasBlobPublishAttempt", "warning");
+
+    DB::Cas::BlobSource source;
+    source.size = payload.size();
+    source.open = [payload] { return std::make_unique<DB::ReadBufferFromOwnString>(payload); };
+    build->putBlob(DB::Cas::tests::idOf(payload), std::move(source));
+
+    EXPECT_EQ(capture.count("failed, attempt 1 of"), 0u) << capture.captured();
+    EXPECT_EQ(capture.count("failed, attempt 2 of"), 1u) << capture.captured();
+}
+
 /// A lost staged `CopyObject` is not resent: the staged copy is spent, so the retry streams.
 TEST(CASBlobPublishTransport, LostStagedCopyIsRepublishedByStreaming)
 {