Merge remote-tracking branch 'origin/topic/awelzel/4754-double-wrapped-broker-data-records'

* origin/topic/awelzel/4754-double-wrapped-broker-data-records:
  cluster/serializer/broker: Do not special case Broker::Data anymore
  broker/Data: Support unwrapping Broker::Data records

CHANGES

@@ -1,3 +1,21 @@
+8.1.0-dev.71 | 2025-08-17 16:56:57 +0200
+
+  * cluster/serializer/broker: Do not special case Broker::Data anymore (Arne Welzel, Corelight)
+
+    The previous approach ignored the fact that nested / inner values might
+    also be Broker::Data values. I'm not super sure about the validity of
+    the test, because it's essentially demonstrating any-nesting, but
+    it's not leading to extra Broker::Data encoding.
+
+  * broker/Data: Support unwrapping Broker::Data records (Arne Welzel, Corelight)
+
+    Calling val_to_data() on a Broker::Data ends up wrapping the
+    Broker::Data record instead of using the contained broker::value
+    directly.
+
+    Seems this should be the default behavior and wonder if the flag
+    even makes sense, but for a 8.0 backport that seems more reasonable.
+
 8.1.0-dev.68 | 2025-08-15 15:20:36 -0700
 
   * Revert "Move BinPAC, bifcl, af_packet, and gen_zam submodules into main zeek repo" (Tim Wojtulewicz)

VERSION

@@ -1 +1 @@
-8.1.0-dev.68
+8.1.0-dev.71

src/broker/Data.cc

@@ -12,6 +12,7 @@
 #include "zeek/IntrusivePtr.h"
 #include "zeek/RE.h"
 #include "zeek/Scope.h"
+#include "zeek/Type.h"
 #include "zeek/broker/data.bif.h"
 #include "zeek/module_util.h"
 
@@ -718,7 +719,7 @@ ValPtr data_to_val(broker::data& d, Type* type) {
     return visit(val_converter{type}, d);
 }
 
-std::optional<broker::data> val_to_data(const Val* v) {
+std::optional<broker::data> val_to_data(const Val* v, bool unwrap_broker_data) {
     switch ( v->GetType()->Tag() ) {
         case TYPE_BOOL: return {v->AsBool()};
         case TYPE_INT: return {v->AsInt()};
@@ -804,7 +805,7 @@ std::optional<broker::data> val_to_data(const Val* v) {
                 composite_key.reserve(vl->Length());
 
                 for ( auto k = 0; k < vl->Length(); ++k ) {
-                    auto key_part = val_to_data(vl->Idx(k).get());
+                    auto key_part = val_to_data(vl->Idx(k).get(), unwrap_broker_data);
 
                     if ( ! key_part )
                         return std::nullopt;
@@ -822,7 +823,7 @@ std::optional<broker::data> val_to_data(const Val* v) {
                 if ( is_set )
                     get<broker::set>(rval).emplace(std::move(key));
                 else {
-                    auto val = val_to_data(te.value->GetVal().get());
+                    auto val = val_to_data(te.value->GetVal().get(), unwrap_broker_data);
 
                     if ( ! val )
                         return std::nullopt;
@@ -846,7 +847,7 @@ std::optional<broker::data> val_to_data(const Val* v) {
                     return std::nullopt;
                 }
 
-                auto item = val_to_data(item_val.get());
+                auto item = val_to_data(item_val.get(), unwrap_broker_data);
 
                 if ( ! item )
                     return std::nullopt;
@@ -871,7 +872,7 @@ std::optional<broker::data> val_to_data(const Val* v) {
                     return std::nullopt;
                 }
 
-                auto item = val_to_data(item_val.get());
+                auto item = val_to_data(item_val.get(), unwrap_broker_data);
 
                 if ( ! item )
                     return std::nullopt;
@@ -883,6 +884,21 @@ std::optional<broker::data> val_to_data(const Val* v) {
         }
         case TYPE_RECORD: {
             auto rec = v->AsRecordVal();
+
+            // If unwrap_broker_data is set and this record is a Broker::Data record,
+            // use the contained data field directly.
+            if ( unwrap_broker_data && rec->GetType() == BifType::Record::Broker::Data ) {
+                const auto ov = rec->GetField<zeek::OpaqueVal>(0);
+                // Sanity.
+                if ( ov->GetType() != opaque_of_data_type ) {
+                    reporter->Error("Broker::Data data field has wrong type: %s",
+                                    obj_desc_short(ov->GetType()).c_str());
+                    return std::nullopt;
+                }
+
+                return static_cast<const DataVal*>(ov.get())->data;
+            }
+
             broker::vector rval;
             size_t num_fields = v->GetType()->AsRecordType()->NumFields();
             rval.reserve(num_fields);
@@ -895,7 +911,7 @@ std::optional<broker::data> val_to_data(const Val* v) {
                     continue;
                 }
 
-                auto item = val_to_data(item_val.get());
+                auto item = val_to_data(item_val.get(), unwrap_broker_data);
 
                 if ( ! item )
                     return std::nullopt;

src/broker/Data.h

@@ -76,9 +76,10 @@ EnumValPtr get_data_type(RecordVal* v, zeek::detail::Frame* frame);
 /**
  * Convert a Zeek value to a Broker data value.
  * @param v a Zeek value.
+ * @param unwrap_broker_data If v or any nested value is a Broker::Data record, use its data broker::value directly.
  * @return a Broker data value if the Zeek value could be converted to one.
  */
-std::optional<broker::data> val_to_data(const Val* v);
+std::optional<broker::data> val_to_data(const Val* v, bool unwrap_broker_data = false);
 
 /**
  * Convert a Broker data value to a Zeek value.

src/cluster/serializer/broker/Serializer.cc

@@ -59,19 +59,10 @@ std::optional<broker::zeek::Event> detail::to_broker_event(const zeek::cluster::
     xs.reserve(ev.Args().size());
 
     for ( const auto& a : ev.Args() ) {
-        if ( a->GetType() == zeek::BifType::Record::Broker::Data ) {
+        if ( auto res = zeek::Broker::detail::val_to_data(a.get(), /*flatten_broker_dataval=*/true) )
-            // When encountering a Broker::Data instance within args, pick out
-            // the broker::data directly to avoid double encoding, Broker::Data.
-            const auto& val = a->AsRecordVal()->GetField(0);
-            auto* data_val = static_cast<zeek::Broker::detail::DataVal*>(val.get());
-            xs.emplace_back(data_val->data);
-        }
-        else if ( auto res = zeek::Broker::detail::val_to_data(a.get()) ) {
             xs.emplace_back(std::move(res.value()));
-        }
+        else
-        else {
             return std::nullopt;
-        }
     }
 
     // Convert metadata from the cluster::detail::Event event to broker's event metadata format.

testing/btest/Baseline/cluster.generic.publish-any/..manager..stdout

@@ -35,3 +35,10 @@ got pong, 27, with, 4, time (cluster publish), Broker::Data, [data=broker::data{
 got pong, 28, with, 4, time (cluster event ), Broker::Data, [data=broker::data{42000000000ns}]
 got pong, 29, with, 4, time (cluster publish), Broker::Data, [data=broker::data{42000000000ns}]
 got pong, 30, with, 4, time (cluster event ), Broker::Data, [data=broker::data{42000000000ns}]
+sending pings, 5, R, [c=42, a=[[c=42, a=hello]]]
+got pong, 31, with, 5, R (cluster publish), Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got pong, 32, with, 5, R (cluster event ), Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got pong, 33, with, 5, R (cluster publish), Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got pong, 34, with, 5, R (cluster event ), Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got pong, 35, with, 5, R (cluster publish), Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got pong, 36, with, 5, R (cluster event ), Broker::Data, [data=broker::data{(42, ((42, hello)))}]

testing/btest/Baseline/cluster.generic.publish-any/..worker-1..stdout

@@ -14,4 +14,7 @@ got ping, 3, vector of count, Broker::Data, [data=broker::data{(1, 2, 3)}]
 got ping, 4, time, Broker::Data, [data=broker::data{42000000000ns}]
 got ping, 4, time, Broker::Data, [data=broker::data{42000000000ns}]
 got ping, 4, time, Broker::Data, [data=broker::data{42000000000ns}]
+got ping, 5, R, Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got ping, 5, R, Broker::Data, [data=broker::data{(42, ((42, hello)))}]
+got ping, 5, R, Broker::Data, [data=broker::data{(42, ((42, hello)))}]
 got finish!

testing/btest/cluster/generic/publish-any.zeek

@@ -34,9 +34,14 @@ global pong: event(c: count, what: string, val: any) &is_used;
 global i = 0;
 global pongs = 0;
 
+type R: record {
+	c: count;
+	a: any;
+};
+
 event send_any()
 	{
-	if ( i > 4 )
+	if ( i > 5 )
 		return;
 
 	local val: any;
@@ -48,8 +53,10 @@ event send_any()
 		val = 42/tcp;
 	else if ( i == 3 )
 		val = vector(1, 2, 3);
-	else
+	else if ( i == 4 )
 		val = double_to_time(42.0);
+	else
+		val = R($c=42, $a=vector(R($c=42, $a="hello")));
 
 	print "sending pings", i, type_name(val), val;
 	Cluster::publish_hrw(Cluster::worker_pool, cat(i), ping, i, type_name(val), val);
@@ -64,10 +71,10 @@ event pong(c: count, what: string, val: any)
 	++pongs;
 	print "got pong", pongs, "with", c, what, type_name(val), val;
 
-	# The manager sends 5 types of pings, in 3 different ways. The worker
+	# The manager sends 6 types of pings, in 3 different ways. The worker
-	# answers each ping in two ways, for a total of 30 expected pongs at the
+	# answers each ping in two ways, for a total of 36 expected pongs at the
 	# manager. Every batch of pings involves 6 pongs.
-	if ( pongs == 30 )
+	if ( pongs == 36 )
 		Cluster::publish(Cluster::worker_topic, finish);
 	else if ( pongs > 0 && pongs % 6  == 0 )
 		{