feat(pipeline): add span_events bindings

Add `addSpanEvent` to append OpenTelemetry-style span events onto the
top-level v0.4 `span_events` field that libdatadog already serializes.
Like meta_struct there is no change-buffer opcode, so the event is
appended directly to the span after draining the queue (span_events do
not depend on any other queued op, so bypassing queue ordering is safe).

Attributes arrive as a flat little-endian buffer with per-value type
tags (String=0, Boolean=1, Integer=2, Double=3, Array=4) matching
libdatadog's AttributeArrayValue discriminants; every read is bounded
against the buffer so a malformed/truncated buffer errors instead of
panicking. A `getSpanEventsJson` helper serializes events via the same
serde impl used for the msgpack wire format, exercised by new
round-trip tests covering each scalar type, arrays, and bounds.

Author: bengl

Cargo.lock

@@ -12,6 +12,7 @@ wasm-bindgen = "0.2"
 wasm-bindgen-futures = "0.4"
 js-sys = "0.3"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1"
 libdatadog-nodejs-capabilities = { path = "../capabilities" }
 libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", branch = "main" }
 libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", branch = "main", default-features = false }

crates/pipeline/Cargo.toml

@@ -18,6 +18,9 @@ use trace_data::*;
 mod stats;
 
 use libdd_trace_utils::change_buffer::{ChangeBuffer, ChangeBufferState};
+use libdd_trace_utils::span::v04::{AttributeAnyValue, AttributeArrayValue, SpanEvent};
+use span_string::SpanString;
+use std::collections::HashMap;
 
 mod utils;
 use utils::*;
@@ -27,6 +30,112 @@ fn init() {
     console_error_panic_hook::set_once();
 }
 
+// --- span event attribute decoding ---
+//
+// `addSpanEvent` receives its attributes as a flat little-endian buffer built
+// by dd-trace-js. Layout: repeated entries until the buffer is exhausted, each
+//   [key_len: u32][key: utf8][tag: u8] + value
+// where the value depends on `tag`:
+//   0 String  [len: u32][utf8]
+//   1 Boolean [u8 (0/1)]
+//   2 Integer [i64]
+//   3 Double  [f64]
+//   4 Array   [count: u32] then `count` items, each [item_tag: u8][scalar]
+//             (item_tag must be 0..=3; nested arrays are rejected)
+// The tags mirror libdatadog's `AttributeArrayValue` discriminants
+// (String=0, Boolean=1, Integer=2, Double=3, Array=4). Every read is bounded
+// against the buffer so a malformed/truncated buffer errors instead of
+// panicking (matching the hardening in `stringTableInsertMany`/`prepareChunk`).
+
+fn se_need(buf: &[u8], idx: usize, n: usize) -> Result<(), JsValue> {
+    // Avoid `idx + n` overflowing: on wasm32 `usize` is 32-bit, and `n` can be
+    // a u32-derived length (e.g. a crafted `key_len`) near `usize::MAX`, which
+    // would wrap and let a too-large read slip past the bound and trap on the
+    // slice. `idx` never exceeds `buf.len()` (it only advances after a checked
+    // read), so `buf.len() - idx` is the safe remaining-byte form.
+    if idx > buf.len() || n > buf.len() - idx {
+        return Err(JsValue::from_str(
+            "addSpanEvent: truncated span-event attribute buffer",
+        ));
+    }
+    Ok(())
+}
+
+fn se_read_u8(buf: &[u8], idx: &mut usize) -> Result<u8, JsValue> {
+    se_need(buf, *idx, 1)?;
+    let b = buf[*idx];
+    *idx += 1;
+    Ok(b)
+}
+
+fn se_read_u32(buf: &[u8], idx: &mut usize) -> Result<u32, JsValue> {
+    se_need(buf, *idx, 4)?;
+    Ok(get_num(buf, idx))
+}
+
+fn se_read_str(buf: &[u8], idx: &mut usize) -> Result<SpanString, JsValue> {
+    let len = se_read_u32(buf, idx)? as usize;
+    se_need(buf, *idx, len)?;
+    let s = std::str::from_utf8(&buf[*idx..*idx + len])
+        .map_err(|e| JsValue::from_str(&format!("addSpanEvent: invalid utf8: {e}")))?;
+    *idx += len;
+    Ok(s.into())
+}
+
+fn se_read_scalar(
+    buf: &[u8],
+    idx: &mut usize,
+    tag: u8,
+) -> Result<AttributeArrayValue<WasmTraceData>, JsValue> {
+    match tag {
+        0 => Ok(AttributeArrayValue::String(se_read_str(buf, idx)?)),
+        1 => Ok(AttributeArrayValue::Boolean(se_read_u8(buf, idx)? != 0)),
+        2 => {
+            se_need(buf, *idx, 8)?;
+            Ok(AttributeArrayValue::Integer(get_num(buf, idx)))
+        }
+        3 => {
+            se_need(buf, *idx, 8)?;
+            Ok(AttributeArrayValue::Double(get_num(buf, idx)))
+        }
+        _ => Err(JsValue::from_str(
+            "addSpanEvent: invalid span-event attribute tag",
+        )),
+    }
+}
+
+fn decode_span_event_attributes(
+    buf: &[u8],
+) -> Result<HashMap<SpanString, AttributeAnyValue<WasmTraceData>>, JsValue> {
+    let mut attributes = HashMap::new();
+    let mut idx = 0usize;
+    while idx < buf.len() {
+        let key = se_read_str(buf, &mut idx)?;
+        let tag = se_read_u8(buf, &mut idx)?;
+        let value = if tag == 4 {
+            let count = se_read_u32(buf, &mut idx)? as usize;
+            // Each item is at least 1 byte (its tag), so cap the pre-allocation
+            // to the remaining buffer: an inflated count can't force a huge
+            // allocation, and the per-item bounded reads catch truncation.
+            let mut items = Vec::with_capacity(count.min(buf.len().saturating_sub(idx)));
+            for _ in 0..count {
+                let item_tag = se_read_u8(buf, &mut idx)?;
+                if item_tag == 4 {
+                    return Err(JsValue::from_str(
+                        "addSpanEvent: nested arrays are not supported",
+                    ));
+                }
+                items.push(se_read_scalar(buf, &mut idx, item_tag)?);
+            }
+            AttributeAnyValue::Array(items)
+        } else {
+            AttributeAnyValue::SingleValue(se_read_scalar(buf, &mut idx, tag)?)
+        };
+        attributes.insert(key, value);
+    }
+    Ok(attributes)
+}
+
 #[wasm_bindgen]
 /// All mutable state is behind RefCell to allow `&self` methods on the
 /// wasm-bindgen wrapper. This prevents re-entrant borrow panics when:
@@ -501,6 +610,51 @@ impl WasmSpanState {
             .unwrap_or(JsValue::NULL))
     }
 
+    // Span events (OpenTelemetry-style) are serialized by libdatadog as the
+    // top-level v0.4 `span_events` field when present. Like meta_struct there
+    // is no change-buffer opcode, so the event is appended directly to the span
+    // after draining the queue (span_events do not depend on any other queued
+    // op, so bypassing queue ordering is safe). `attrs_buf` is the flat typed
+    // attribute encoding decoded by `decode_span_event_attributes`.
+    #[wasm_bindgen(js_name = "addSpanEvent")]
+    pub fn add_span_event(
+        &self,
+        span_id: u64,
+        name: &str,
+        time_unix_nano: u64,
+        attrs_buf: &[u8],
+    ) -> Result<(), JsValue> {
+        self.flush_change_queue()?;
+        // Decode before borrowing cbs mutably so a malformed buffer errors
+        // without holding the borrow.
+        let attributes = decode_span_event_attributes(attrs_buf)?;
+        let mut cbs = self.cbs.borrow_mut();
+        let span = cbs
+            .span_mut(span_id)
+            .map_err(|e| JsValue::from_str(&e.to_string()))?;
+        span.span_events.push(SpanEvent {
+            time_unix_nano,
+            name: name.into(),
+            attributes,
+        });
+        Ok(())
+    }
+
+    // Test/inspection helper: serialize the span's events to JSON via the same
+    // serde `Serialize` impl libdatadog uses for the msgpack wire format, so
+    // the `type`/`*_value` shape mirrors exactly what is sent to the agent
+    // (String=0, Boolean=1, Integer=2, Double=3, Array=4).
+    #[wasm_bindgen(js_name = "getSpanEventsJson")]
+    pub fn get_span_events_json(&self, span_id: u64) -> Result<String, JsValue> {
+        self.flush_change_queue()?;
+        let cbs = self.cbs.borrow();
+        let span = cbs
+            .get_span(span_id)
+            .map_err(|e| JsValue::from_str(&e.to_string()))?;
+        serde_json::to_string(&span.span_events)
+            .map_err(|e| JsValue::from_str(&format!("getSpanEventsJson: {e}")))
+    }
+
     // Trace-level attributes live on the Segment (keyed by segment_id, which
     // JS allocates and shares across spans in the same local trace).
     #[wasm_bindgen(js_name = "getTraceMetaAttr")]

crates/pipeline/src/lib.rs

@@ -1,9 +1,13 @@
 use libdd_trace_utils::span::TraceData;
+use serde::Serialize;
 
 use crate::span_bytes::SpanBytesImpl;
 use crate::span_string::SpanString;
 
-#[derive(Clone, Default, Debug, PartialEq)]
+// `Serialize` is derived only so the test helper `getSpanEventsJson` can
+// serialize `Vec<SpanEvent<WasmTraceData>>` (serde's derive on the generic
+// `SpanEvent<T>` requires `T: Serialize`). The unit struct carries no data.
+#[derive(Clone, Default, Debug, PartialEq, Serialize)]
 pub struct WasmTraceData;
 
 impl TraceData for WasmTraceData {

crates/pipeline/src/trace_data.rs

@@ -90,6 +90,20 @@ class Span {
     return this.nativeSpans.state.getMetaStruct(this.spanIdBig, key)
   }
 
+  addSpanEvent (name, timeUnixNano, attributes = {}) {
+    this.nativeSpans.state.addSpanEvent(
+      this.spanIdBig,
+      name,
+      BigInt(timeUnixNano),
+      encodeSpanEventAttrs(attributes)
+    )
+    return this
+  }
+
+  getSpanEvents () {
+    return JSON.parse(this.nativeSpans.state.getSpanEventsJson(this.spanIdBig))
+  }
+
   finish () {
     this.duration = BigInt(Date.now()) * 1000000n - this._startTime
     return this
@@ -296,6 +310,41 @@ class NativeSpansInterface {
   }
 }
 
+// Build the flat span-event attribute buffer consumed by the Rust decoder
+// (`decode_span_event_attributes` in crates/pipeline/src/lib.rs). This mirrors
+// what dd-trace-js's `addSpanEvent` wrapper produces. Tags: String=0,
+// Boolean=1, Integer=2, Double=3, Array=4 (matching libdatadog's
+// AttributeArrayValue discriminants).
+function encodeSpanEventAttrs (attributes) {
+  const enc = new TextEncoder()
+  const chunks = []
+  const u32 = (n) => { const b = Buffer.alloc(4); b.writeUInt32LE(n >>> 0, 0); return b }
+  const i64 = (n) => { const b = Buffer.alloc(8); b.writeBigInt64LE(BigInt(n), 0); return b }
+  const f64 = (n) => { const b = Buffer.alloc(8); b.writeDoubleLE(n, 0); return b }
+  const str = (s) => { const sb = Buffer.from(enc.encode(s)); return Buffer.concat([u32(sb.length), sb]) }
+  // Returns `[tag][value]` — used both for single values and array items.
+  const scalar = (v) => {
+    if (typeof v === 'string') return Buffer.concat([Buffer.from([0]), str(v)])
+    if (typeof v === 'boolean') return Buffer.concat([Buffer.from([1]), Buffer.from([v ? 1 : 0])])
+    if (typeof v === 'number') {
+      return Number.isInteger(v)
+        ? Buffer.concat([Buffer.from([2]), i64(v)])
+        : Buffer.concat([Buffer.from([3]), f64(v)])
+    }
+    throw new TypeError(`unsupported span-event attribute value: ${typeof v}`)
+  }
+  for (const [key, value] of Object.entries(attributes)) {
+    chunks.push(str(key))
+    if (Array.isArray(value)) {
+      chunks.push(Buffer.from([4]), u32(value.length))
+      for (const item of value) chunks.push(scalar(item))
+    } else {
+      chunks.push(scalar(value))
+    }
+  }
+  return new Uint8Array(Buffer.concat(chunks))
+}
+
 describe('pipeline', () => {
   let nativeSpans
 
@@ -414,6 +463,93 @@ describe('pipeline', () => {
     })
   })
 
+  describe('span_events', () => {
+    it('appends an event with no attributes', () => {
+      const span = nativeSpans.createSpan()
+      span.addSpanEvent('exception', 1727211691770716000n)
+
+      const events = span.getSpanEvents()
+      assert.strictEqual(events.length, 1)
+      assert.strictEqual(events[0].name, 'exception')
+      assert.strictEqual(events[0].time_unix_nano, 1727211691770716000)
+      // Empty attributes are skipped by libdatadog's serializer.
+      assert.strictEqual(events[0].attributes, undefined)
+    })
+
+    it('round-trips scalar attributes of every type with correct type tags', () => {
+      const span = nativeSpans.createSpan()
+      span.addSpanEvent('evt', 1000n, {
+        s: 'hello',
+        b: true,
+        i: 42,
+        d: 3.5
+      })
+
+      const [event] = span.getSpanEvents()
+      assert.strictEqual(event.name, 'evt')
+      assert.strictEqual(event.time_unix_nano, 1000)
+      // type tags: String=0, Boolean=1, Integer=2, Double=3
+      assert.deepStrictEqual(event.attributes.s, { type: 0, string_value: 'hello' })
+      assert.deepStrictEqual(event.attributes.b, { type: 1, bool_value: true })
+      assert.deepStrictEqual(event.attributes.i, { type: 2, int_value: 42 })
+      assert.deepStrictEqual(event.attributes.d, { type: 3, double_value: 3.5 })
+    })
+
+    it('round-trips an array attribute (type 4) with typed items', () => {
+      const span = nativeSpans.createSpan()
+      span.addSpanEvent('evt', 1n, { tags: ['a', 'b'], nums: [1, 2, 3] })
+
+      const [event] = span.getSpanEvents()
+      assert.deepStrictEqual(event.attributes.tags, {
+        type: 4,
+        array_value: { values: [{ type: 0, string_value: 'a' }, { type: 0, string_value: 'b' }] }
+      })
+      assert.deepStrictEqual(event.attributes.nums, {
+        type: 4,
+        array_value: { values: [{ type: 2, int_value: 1 }, { type: 2, int_value: 2 }, { type: 2, int_value: 3 }] }
+      })
+    })
+
+    it('appends multiple events in order', () => {
+      const span = nativeSpans.createSpan()
+      span.addSpanEvent('first', 1n)
+      span.addSpanEvent('second', 2n, { k: 'v' })
+
+      const events = span.getSpanEvents()
+      assert.strictEqual(events.length, 2)
+      assert.strictEqual(events[0].name, 'first')
+      assert.strictEqual(events[1].name, 'second')
+      assert.deepStrictEqual(events[1].attributes.k, { type: 0, string_value: 'v' })
+    })
+
+    it('returns an empty array for a span with no events', () => {
+      const span = nativeSpans.createSpan()
+      assert.deepStrictEqual(span.getSpanEvents(), [])
+    })
+
+    it('rejects a truncated attribute buffer instead of panicking', () => {
+      const span = nativeSpans.createSpan()
+      // key_len=5 but no key bytes follow → bounded read must error.
+      const bad = new Uint8Array([5, 0, 0, 0])
+      assert.throws(
+        () => span.nativeSpans.state.addSpanEvent(span.spanIdBig, 'evt', 1n, bad),
+        /truncated span-event attribute buffer/
+      )
+    })
+
+    it('rejects an overflowing key_len without trapping (wasm32 usize)', () => {
+      const span = nativeSpans.createSpan()
+      // key_len = 0xFFFFFFFF: on wasm32 `idx + key_len` would wrap and slip
+      // past the bound, trapping on the slice. The remaining-byte form must
+      // reject it as a truncated buffer instead.
+      const bad = new Uint8Array([0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00])
+      assert.throws(
+        () => span.nativeSpans.state.addSpanEvent(span.spanIdBig, 'evt', 1n, bad),
+        /truncated span-event attribute buffer/
+      )
+    })
+  })
+
   describe('span timing', () => {
     it('should set and get start time', () => {
       const span = nativeSpans.createSpan()