More graceful reporting of Protobuf protocol decode errors

If a message fails decoding due to e.g. a missing document type
during document decode, that will be reported as a protocol-level
decode error, even if that message itself is perfectly compliant.
Report such exceptions as warnings instead, but add more context
to the exception message to make it clear that it happened in a
decode-context.

Example of old (and rather confusing) message in logs:
```
WARNING distributor  vds.documentprotocol  Document type foo not found
```
Now becomes:
```
WARNING distributor  vds.documentprotocol Failed decoding message of
  type PutDocumentRequest: Document type foo not found
```

2 files changed, 15 insertions, 9 deletions

diff --git a/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/RoutableFactories80.java b/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/RoutableFactories80.java
index 4712f6d2442..ef1575e2101 100644
--- a/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/RoutableFactories80.java
+++ b/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/RoutableFactories80.java
@@ -73,7 +73,7 @@ abstract class RoutableFactories80 {
                 protoStream.checkNoSpaceLeft();
                 return buf;
             } catch (IOException | RuntimeException e) {
-                logCodecError("encoding", e);
+                log.severe("Error during Protobuf encoding of message type %s: %s".formatted(apiClass.getSimpleName(), e.getMessage()));
                 return null;
             }
         }
@@ -89,17 +89,11 @@ abstract class RoutableFactories80 {
             try {
                 return decoderFn.apply(in);
             } catch (RuntimeException e) {
-                logCodecError("decoding", e);
-                return null;
+                throw new RuntimeException("Error during Protobuf decoding of message type %s: %s"
+                        .formatted(apiClass.getSimpleName(), e.getMessage()), e);
             }
         }
 
-        private void logCodecError(String op, Throwable t) {
-            // TODO ideally we'd print the peer address here, but that information is Abstracted Away(tm)
-            //  and we can't safely propagate the exception to a lower-level component which _does_ know this
-            //  information, as it's not necessarily exception safe.
-            log.severe("Error during Protobuf %s for message type %s: %s".formatted(op, apiClass.getSimpleName(), t.getMessage()));
-        }
     }
 
     private static class ProtobufCodecBuilder<DocApiT extends Routable, ProtoT extends AbstractMessage> {
diff --git a/documentapi/src/vespa/documentapi/messagebus/routable_factories_8.cpp b/documentapi/src/vespa/documentapi/messagebus/routable_factories_8.cpp
index c707d8bb9fc..f9782e1abd9 100644
--- a/documentapi/src/vespa/documentapi/messagebus/routable_factories_8.cpp
+++ b/documentapi/src/vespa/documentapi/messagebus/routable_factories_8.cpp
@@ -11,6 +11,7 @@
 #include <vespa/documentapi/messagebus/docapi_visiting.pb.h>
 #include <vespa/documentapi/messagebus/docapi_inspect.pb.h>
 #include <vespa/vespalib/objects/nbostream.h>
+#include <vespa/vespalib/util/exceptions.h>
 #include <vespa/vespalib/util/stringfmt.h>
 
 #include <vespa/log/bufferedlogger.h>
@@ -147,6 +148,11 @@ void log_codec_error(const char* op, const char* type, const char* msg) noexcept
     LOGBM(error, "Error during Protobuf %s for message type %s: %s", op, type, msg);
 }
 
+[[noreturn]] void rethrow_as_decorated_exception(const char* type, const vespalib::string& msg) __attribute((noinline));
+[[noreturn]] void rethrow_as_decorated_exception(const char* type, const vespalib::string& msg) {
+    throw vespalib::IllegalArgumentException(vespalib::make_string("Failed decoding message of type %s: %s", type, msg.c_str()), VESPA_STRLOC);
+}
+
 template <typename DocApiType, typename ProtobufType, typename EncodeFn, typename DecodeFn>
 requires std::is_invocable_r_v<void, EncodeFn, const DocApiType&, ProtobufType&> &&
          std::is_invocable_r_v<std::unique_ptr<DocApiType>, DecodeFn, const ProtobufType&>
@@ -193,6 +199,12 @@ public:
                 msg->setApproxSize(buf_size); // Wire size is a proxy for in-memory size
             }
             return msg;
+        } catch (vespalib::Exception& e) {
+            // We handle vespalib exceptions more "gently" (by propagating them up to the DocumentProtocol)
+            // in that we assume they are caused by transient problems such as missing document types etc.,
+            // rather than indicating a more serious error. But to make debugging easier we add some
+            // explicit context to the exception message and throw a new exception in its place.
+            rethrow_as_decorated_exception(ProtobufType::descriptor()->name().c_str(), e.getMessage());
         } catch (std::exception& e) {
             log_codec_error("decode", ProtobufType::descriptor()->name().c_str(), e.what());
             return {};