Repository: honeycombio/refinery
Pull request #1613 (2025-07-03 19:25:23)

maint: clean up payload code and testing, fix broken benchmark
## Which problem is this PR solving?
All our metadata warmup code wasn't 100% baked when landed.

## Short description of the changes
Restructures the big `metadataFields` table to make it much more compact.

Makes `metadataFields` a map for (hopefully) faster lookups when checking a single field by key, which seems important now that this is a longer list of fields.

More consistently ignores metadata field values which are in the `memoizedFields` map, and prevents them being added via `Set`. This means you can no longer manually set a meta field in the payload to a value of the wrong type, nor will we blindly copy these values from input to output during serialization. Probably won't come up in this codebase but we'll be better off with a single source of truth on these values.

Fixes broken BenchmarkRouterBatch

Cleans up some redundant test code and adds unmarshal testing for all metadata fields in Payload.



Patch:
diff --git a/route/route.go b/route/route.go
index b7e8f53..b3c3443 100644
--- a/route/route.go
+++ b/route/route.go
@@ -385,22 +385,16 @@ func (r *Router) marshalToFormat(w http.ResponseWriter, obj interface{}, format
 // event is handler for /1/event/
 func (r *Router) event(w http.ResponseWriter, req *http.Request) {
 	r.Metrics.Increment(r.incomingOrPeer + "_router_event")
-	defer req.Body.Close()
 
 	ctx := req.Context()
-	bodyReader, err := r.getMaybeCompressedBody(req)
-	if err != nil {
-		r.handlerReturnWithError(w, ErrPostBody, err)
-		return
-	}
-
-	reqBod, err := io.ReadAll(bodyReader)
+	bodyBuffer, err := r.readAndCloseMaybeCompressedBody(req)
 	if err != nil {
 		r.handlerReturnWithError(w, ErrPostBody, err)
 		return
 	}
+	defer recycleHTTPBodyBuffer(bodyBuffer)
 
-	ev, err := r.requestToEvent(ctx, req, reqBod)
+	ev, err := r.requestToEvent(ctx, req, bodyBuffer.Bytes())
 	if err != nil {
 		r.handlerReturnWithError(w, ErrReqToEvent, err)
 		return
@@ -459,30 +453,22 @@ func (r *Router) requestToEvent(ctx context.Context, req *http.Request, reqBod [
 
 func (r *Router) batch(w http.ResponseWriter, req *http.Request) {
 	r.Metrics.Increment(r.incomingOrPeer + "_router_batch")
-	defer req.Body.Close()
 
 	ctx := req.Context()
 	reqID := ctx.Value(types.RequestIDContextKey{})
 	debugLog := r.iopLogger.Debug().WithField("request_id", reqID)
 
-	bodyReader, err := r.getMaybeCompressedBody(req)
-	if err != nil {
-		r.handlerReturnWithError(w, ErrPostBody, err)
-		return
-	}
-
-	reqBod, err := io.ReadAll(bodyReader)
+	bodyBuffer, err := r.readAndCloseMaybeCompressedBody(req)
 	if err != nil {
 		r.handlerReturnWithError(w, ErrPostBody, err)
 		return
 	}
+	defer recycleHTTPBodyBuffer(bodyBuffer)
 
 	batchedEvents := newBatchedEvents(r.Config)
-
-	err = unmarshal(req, reqBod, batchedEvents)
-
+	err = unmarshal(req, bodyBuffer.Bytes(), batchedEvents)
 	if err != nil {
-		debugLog.WithField("error", err.Error()).WithField("request.url", req.URL).WithField("json_body", string(reqBod)).Logf("error parsing json")
+		debugLog.WithField("error", err.Error()).WithField("request.url", req.URL).WithField("json_body", string(bodyBuffer.Bytes())).Logf("error parsing json")
 		r.handlerReturnWithError(w, ErrJSONFailed, err)
 		return
 	}
@@ -702,7 +688,28 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 	return nil
 }
 
-func (r *Router) getMaybeCompressedBody(req *http.Request) (io.Reader, error) {
+// A pool of buffers for HTTP bodies; there's no sorting by size here, since
+// the assumption is that over time these will converge on the largest typical
+// body size used in this environment.
+var httpBodyBufferPool = sync.Pool{
+	New: func() any {
+		return new(bytes.Buffer)
+	},
+}
+
+func recycleHTTPBodyBuffer(b *bytes.Buffer) {
+	if b != nil {
+		b.Reset()
+		httpBodyBufferPool.Put(b)
+	}
+}
+
+// When finished with the returned buffer, callers should return it to the pool
+// by calling recycleHTTPBodyBuffer.
+// Reads the body and immediately closes it, releasing resources as soon as possible.
+func (r *Router) readAndCloseMaybeCompressedBody(req *http.Request) (*bytes.Buffer, error) {
+	defer req.Body.Close()
+
 	var reader io.Reader
 	switch req.Header.Get("Content-Encoding") {
 	case "gzip":
@@ -711,12 +718,7 @@ func (r *Router) getMaybeCompressedBody(req *http.Request) (io.Reader, error) {
 			return nil, err
 		}
 		defer gzipReader.Close()
-
-		buf := &bytes.Buffer{}
-		if _, err := io.Copy(buf, io.LimitReader(gzipReader, HTTPMessageSizeMax)); err != nil {
-			return nil, err
-		}
-		reader = buf
+		reader = gzipReader
 	case "zstd":
 		zReader := <-r.zstdDecoders
 		defer func(zReader *zstd.Decoder) {
@@ -728,16 +730,18 @@ func (r *Router) getMaybeCompressedBody(req *http.Request) (io.Reader, error) {
 		if err != nil {
 			return nil, err
 		}
-		buf := &bytes.Buffer{}
-		if _, err := io.Copy(buf, io.LimitReader(zReader, HTTPMessageSizeMax)); err != nil {
-			return nil, err
-		}
-
-		reader = buf
+		reader = zReader
 	default:
 		reader = req.Body
 	}
-	return reader, nil
+
+	buf := httpBodyBufferPool.Get().(*bytes.Buffer)
+	if _, err := io.Copy(buf, io.LimitReader(reader, HTTPMessageSizeMax)); err != nil {
+		recycleHTTPBodyBuffer(buf)
+		return nil, err
+	}
+
+	return buf, nil
 }
 
 // getEventTime tries to guess the time format in our time header!
diff --git a/types/payload.go b/types/payload.go
index 4c977f5..6db89e4 100644
--- a/types/payload.go
+++ b/types/payload.go
@@ -42,7 +42,7 @@ const (
 	MetaRefineryForwarded         = "meta.refinery.forwarded"
 	MetaRefineryExpiredTrace      = "meta.refinery.expired_trace"
 
-	//	These fields are not used by the refinery itself for sampling decisions.
+	// These fields are not used by the refinery itself for sampling decisions.
 	// They are used to pass information from refinery to honeycomb.
 	MetaRefineryLocalHostname      = "meta.refinery.local_hostname"
 	MetaStressed                   = "meta.stressed"
@@ -66,7 +66,6 @@ const (
 // could be implemented in terms of get and set but this would transit the
 // concrete values through type any, which is inefficient. This is the compromise.
 type metadataField struct {
-	key           string
 	expectedType  FieldType
 	get           func(p *Payload) (value any, ok bool)               // Payload.Get, Payload.All
 	set           func(p *Payload, value any)                         // Payload.Set
@@ -75,399 +74,96 @@ type metadataField struct {
 	unmarshalMsgp func(p *Payload, in []byte) (out []byte, err error) // Payload.extractMetadataFromBytes
 }
 
-var metadataFields = []metadataField{
-	{
-		key:          MetaSignalType,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaSignalType != "" {
-				return p.MetaSignalType, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaSignalType = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaSignalType != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaSignalType != "" {
-				out = msgp.AppendString(in, MetaSignalType)
-				out = msgp.AppendString(out, p.MetaSignalType)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaSignalType, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaTraceID,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaTraceID != "" {
-				return p.MetaTraceID, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaTraceID = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaTraceID != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaTraceID, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaAnnotationType,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaAnnotationType != "" {
-				return p.MetaAnnotationType, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaAnnotationType = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaAnnotationType != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaAnnotationType != "" {
-				out = msgp.AppendString(in, MetaAnnotationType)
-				out = msgp.AppendString(out, p.MetaAnnotationType)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaAnnotationType, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryProbe,
-		expectedType: FieldTypeBool,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryProbe.HasValue {
-				return p.MetaRefineryProbe.Value, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(bool); ok {
-				p.MetaRefineryProbe.Set(v)
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryProbe.HasValue
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryProbe.HasValue {
-				out = msgp.AppendString(in, MetaRefineryProbe)
-				out = msgp.AppendBool(out, p.MetaRefineryProbe.Value)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			var val bool
-			val, out, err = msgp.ReadBoolBytes(in)
-			if err == nil {
-				p.MetaRefineryProbe.Set(val)
-			}
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryRoot,
-		expectedType: FieldTypeBool,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryRoot.HasValue {
-				return p.MetaRefineryRoot.Value, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(bool); ok {
-				p.MetaRefineryRoot.Set(v)
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryRoot.HasValue
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryRoot.HasValue && p.MetaRefineryRoot.Value {
-				out = msgp.AppendString(in, MetaRefineryRoot)
-				out = msgp.AppendBool(out, p.MetaRefineryRoot.Value)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			var val bool
-			val, out, err = msgp.ReadBoolBytes(in)
-			if err == nil {
-				p.MetaRefineryRoot.Set(val)
-			}
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryIncomingUserAgent,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryIncomingUserAgent != "" {
-				return p.MetaRefineryIncomingUserAgent, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaRefineryIncomingUserAgent = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryIncomingUserAgent != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryIncomingUserAgent != "" {
-				out = msgp.AppendString(in, MetaRefineryIncomingUserAgent)
-				out = msgp.AppendString(out, p.MetaRefineryIncomingUserAgent)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefineryIncomingUserAgent, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefinerySendBy,
-		expectedType: FieldTypeInt64,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefinerySendBy != 0 {
-				return p.MetaRefinerySendBy, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(int64); ok {
-				p.MetaRefinerySendBy = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefinerySendBy != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefinerySendBy != 0 {
-				out = msgp.AppendString(in, MetaRefinerySendBy)
-				out = msgp.AppendInt64(out, p.MetaRefinerySendBy)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefinerySendBy, out, err = msgp.ReadInt64Bytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefinerySpanDataSize,
-		expectedType: FieldTypeInt64,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefinerySpanDataSize != 0 {
-				return p.MetaRefinerySpanDataSize, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(int64); ok {
-				p.MetaRefinerySpanDataSize = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefinerySpanDataSize != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefinerySpanDataSize != 0 {
-				out = msgp.AppendString(in, MetaRefinerySpanDataSize)
-				out = msgp.AppendInt64(out, p.MetaRefinerySpanDataSize)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefinerySpanDataSize, out, err = msgp.ReadInt64Bytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryMinSpan,
-		expectedType: FieldTypeBool,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryMinSpan.HasValue {
-				return p.MetaRefineryMinSpan.Value, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(bool); ok {
-				p.MetaRefineryMinSpan.Set(v)
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryMinSpan.HasValue
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryMinSpan.HasValue {
-				out = msgp.AppendString(in, MetaRefineryMinSpan)
-				out = msgp.AppendBool(out, p.MetaRefineryMinSpan.Value)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			var val bool
-			val, out, err = msgp.ReadBoolBytes(in)
-			if err == nil {
-				p.MetaRefineryMinSpan.Set(val)
-			}
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryForwarded,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryForwarded != "" {
-				return p.MetaRefineryForwarded, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaRefineryForwarded = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryForwarded != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryForwarded != "" {
-				out = msgp.AppendString(in, MetaRefineryForwarded)
-				out = msgp.AppendString(out, p.MetaRefineryForwarded)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefineryForwarded, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryExpiredTrace,
-		expectedType: FieldTypeBool,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryExpiredTrace.HasValue {
-				return p.MetaRefineryExpiredTrace.Value, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(bool); ok {
-				p.MetaRefineryExpiredTrace.Set(v)
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryExpiredTrace.HasValue
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryExpiredTrace.HasValue {
-				out = msgp.AppendString(in, MetaRefineryExpiredTrace)
-				out = msgp.AppendBool(out, p.MetaRefineryExpiredTrace.Value)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			var val bool
-			val, out, err = msgp.ReadBoolBytes(in)
-			if err == nil {
-				p.MetaRefineryExpiredTrace.Set(val)
-			}
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryLocalHostname,
+var metadataFields = map[string]metadataField{
+	MetaSignalType:                stringField(MetaSignalType, func(p *Payload) *string { return &p.MetaSignalType }),
+	MetaTraceID:                   stringField(MetaTraceID, func(p *Payload) *string { return &p.MetaTraceID }),
+	MetaAnnotationType:            stringField(MetaAnnotationType, func(p *Payload) *string { return &p.MetaAnnotationType }),
+	MetaRefineryProbe:             boolField(MetaRefineryProbe, func(p *Payload) *nullableBool { return &p.MetaRefineryProbe }),
+	MetaRefineryRoot:              boolField(MetaRefineryRoot, func(p *Payload) *nullableBool { return &p.MetaRefineryRoot }),
+	MetaRefineryIncomingUserAgent: stringField(MetaRefineryIncomingUserAgent, func(p *Payload) *string { return &p.MetaRefineryIncomingUserAgent }),
+	MetaRefinerySendBy:            int64Field(MetaRefinerySendBy, func(p *Payload) *int64 { return &p.MetaRefinerySendBy }),
+	MetaRefinerySpanDataSize:      int64Field(MetaRefinerySpanDataSize, func(p *Payload) *int64 { return &p.MetaRefinerySpanDataSize }),
+	MetaRefineryMinSpan:           boolField(MetaRefineryMinSpan, func(p *Payload) *nullableBool { return &p.MetaRefineryMinSpan }),
+	MetaRefineryForwarded:         stringField(MetaRefineryForwarded, func(p *Payload) *string { return &p.MetaRefineryForwarded }),
+	MetaRefineryExpiredTrace:      boolField(MetaRefineryExpiredTrace, func(p *Payload) *nullableBool { return &p.MetaRefineryExpiredTrace }),
+
+	MetaRefineryLocalHostname:      stringField(MetaRefineryLocalHostname, func(p *Payload) *string { return &p.MetaRefineryLocalHostname }),
+	MetaStressed:                   boolField(MetaStressed, func(p *Payload) *nullableBool { return &p.MetaStressed }),
+	MetaRefineryReason:             stringField(MetaRefineryReason, func(p *Payload) *string { return &p.MetaRefineryReason }),
+	MetaRefinerySendReason:         stringField(MetaRefinerySendReason, func(p *Payload) *string { return &p.MetaRefinerySendReason }),
+	MetaSpanEventCount:             int64Field(MetaSpanEventCount, func(p *Payload) *int64 { return &p.MetaSpanEventCount }),
+	MetaSpanLinkCount:              int64Field(MetaSpanLinkCount, func(p *Payload) *int64 { return &p.MetaSpanLinkCount }),
+	MetaSpanCount:                  int64Field(MetaSpanCount, func(p *Payload) *int64 { return &p.MetaSpanCount }),
+	MetaEventCount:                 int64Field(MetaEventCount, func(p *Payload) *int64 { return &p.MetaEventCount }),
+	MetaRefineryOriginalSampleRate: int64Field(MetaRefineryOriginalSampleRate, func(p *Payload) *int64 { return &p.MetaRefineryOriginalSampleRate }),
+	MetaRefineryShutdownSend:       boolField(MetaRefineryShutdownSend, func(p *Payload) *nullableBool { return &p.MetaRefineryShutdownSend }),
+	MetaRefinerySampleKey:          stringField(MetaRefinerySampleKey, func(p *Payload) *string { return &p.MetaRefinerySampleKey }),
+}
+
+// Helpers to set up metadataField entries based on the supplied key and
+// ptr function to get access to the correct field.
+func stringField(key string, ptr func(*Payload) *string) metadataField {
+	return metadataField{
 		expectedType: FieldTypeString,
 		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryLocalHostname != "" {
-				return p.MetaRefineryLocalHostname, true
+			strPtr := ptr(p)
+			if *strPtr != "" {
+				return *strPtr, true
 			}
 			return nil, false
 		},
 		set: func(p *Payload, value any) {
 			if v, ok := value.(string); ok {
-				p.MetaRefineryLocalHostname = v
+				strPtr := ptr(p)
+				*strPtr = v
 			}
 		},
 		exist: func(p *Payload) bool {
-			return p.MetaRefineryLocalHostname != ""
+			strPtr := ptr(p)
+			return *strPtr != ""
 		},
 		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryLocalHostname != "" {
-				out = msgp.AppendString(in, MetaRefineryLocalHostname)
-				out = msgp.AppendString(out, p.MetaRefineryLocalHostname)
+			strPtr := ptr(p)
+			if *strPtr != "" {
+				out = msgp.AppendString(in, key)
+				out = msgp.AppendString(out, *strPtr)
 				return out, true
 			}
 			return in, false
 		},
 		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefineryLocalHostname, out, err = msgp.ReadStringBytes(in)
+			strPtr := ptr(p)
+			*strPtr, out, err = msgp.ReadStringBytes(in)
 			return out, err
 		},
-	},
-	{
-		key:          MetaStressed,
+	}
+}
+
+func boolField(key string, ptr func(*Payload) *nullableBool) metadataField {
+	return metadataField{
 		expectedType: FieldTypeBool,
 		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaStressed.HasValue {
-				return p.MetaStressed.Value, true
+			boolPtr := ptr(p)
+			if boolPtr.HasValue {
+				return boolPtr.Value, true
 			}
 			return nil, false
 		},
 		set: func(p *Payload, value any) {
 			if v, ok := value.(bool); ok {
-				p.MetaStressed.Set(v)
+				boolPtr := ptr(p)
+				boolPtr.Set(v)
 			}
 		},
 		exist: func(p *Payload) bool {
-			return p.MetaStressed.HasValue
+			boolPtr := ptr(p)
+			return boolPtr.HasValue
 		},
 		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaStressed.HasValue {
-				out = msgp.AppendString(in, MetaStressed)
-				out = msgp.AppendBool(out, p.MetaStressed.Value)
+			boolPtr := ptr(p)
+			if boolPtr.HasValue {
+				out = msgp.AppendString(in, key)
+				out = msgp.AppendBool(out, boolPtr.Value)
 				return out, true
 			}
 			return in, false
@@ -476,295 +172,49 @@ var metadataFields = []metadataField{
 			var val bool
 			val, out, err = msgp.ReadBoolBytes(in)
 			if err == nil {
-				p.MetaStressed.Set(val)
-			}
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryReason,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryReason != "" {
-				return p.MetaRefineryReason, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaRefineryReason = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryReason != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryReason != "" {
-				out = msgp.AppendString(in, MetaRefineryReason)
-				out = msgp.AppendString(out, p.MetaRefineryReason)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefineryReason, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefinerySendReason,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefinerySendReason != "" {
-				return p.MetaRefinerySendReason, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaRefinerySendReason = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefinerySendReason != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefinerySendReason != "" {
-				out = msgp.AppendString(in, MetaRefinerySendReason)
-				out = msgp.AppendString(out, p.MetaRefinerySendReason)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefinerySendReason, out, err = msgp.ReadStringBytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaSpanEventCount,
-		expectedType: FieldTypeInt64,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaSpanEventCount != 0 {
-				return p.MetaSpanEventCount, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(int64); ok {
-				p.MetaSpanEventCount = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaSpanEventCount != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaSpanEventCount != 0 {
-				out = msgp.AppendString(in, MetaSpanEventCount)
-				out = msgp.AppendInt64(out, p.MetaSpanEventCount)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaSpanEventCount, out, err = msgp.ReadInt64Bytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaSpanLinkCount,
-		expectedType: FieldTypeInt64,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaSpanLinkCount != 0 {
-				return p.MetaSpanLinkCount, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(int64); ok {
-				p.MetaSpanLinkCount = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaSpanLinkCount != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaSpanLinkCount != 0 {
-				out = msgp.AppendString(in, MetaSpanLinkCount)
-				out = msgp.AppendInt64(out, p.MetaSpanLinkCount)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaSpanLinkCount, out, err = msgp.ReadInt64Bytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaSpanCount,
-		expectedType: FieldTypeInt64,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaSpanCount != 0 {
-				return p.MetaSpanCount, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(int64); ok {
-				p.MetaSpanCount = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaSpanCount != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaSpanCount != 0 {
-				out = msgp.AppendString(in, MetaSpanCount)
-				out = msgp.AppendInt64(out, p.MetaSpanCount)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaSpanCount, out, err = msgp.ReadInt64Bytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaEventCount,
-		expectedType: FieldTypeInt64,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaEventCount != 0 {
-				return p.MetaEventCount, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(int64); ok {
-				p.MetaEventCount = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaEventCount != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaEventCount != 0 {
-				out = msgp.AppendString(in, MetaEventCount)
-				out = msgp.AppendInt64(out, p.MetaEventCount)
-				return out, true
+				boolPtr := ptr(p)
+				boolPtr.Set(val)
 			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaEventCount, out, err = msgp.ReadInt64Bytes(in)
 			return out, err
 		},
-	},
-	{
-		key:          MetaRefineryOriginalSampleRate,
+	}
+}
+
+func int64Field(key string, ptr func(*Payload) *int64) metadataField {
+	return metadataField{
 		expectedType: FieldTypeInt64,
 		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryOriginalSampleRate != 0 {
-				return p.MetaRefineryOriginalSampleRate, true
+			intPtr := ptr(p)
+			if *intPtr != 0 {
+				return *intPtr, true
 			}
 			return nil, false
 		},
 		set: func(p *Payload, value any) {
 			if v, ok := value.(int64); ok {
-				p.MetaRefineryOriginalSampleRate = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefineryOriginalSampleRate != 0
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryOriginalSampleRate != 0 {
-				out = msgp.AppendString(in, MetaRefineryOriginalSampleRate)
-				out = msgp.AppendInt64(out, p.MetaRefineryOriginalSampleRate)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefineryOriginalSampleRate, out, err = msgp.ReadInt64Bytes(in)
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefineryShutdownSend,
-		expectedType: FieldTypeBool,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefineryShutdownSend.HasValue {
-				return p.MetaRefineryShutdownSend.Value, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(bool); ok {
-				p.MetaRefineryShutdownSend.Set(v)
+				intPtr := ptr(p)
+				*intPtr = v
 			}
 		},
 		exist: func(p *Payload) bool {
-			return p.MetaRefineryShutdownSend.HasValue
+			intPtr := ptr(p)
+			return *intPtr != 0
 		},
 		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefineryShutdownSend.HasValue {
-				out = msgp.AppendString(in, MetaRefineryShutdownSend)
-				out = msgp.AppendBool(out, p.MetaRefineryShutdownSend.Value)
+			intPtr := ptr(p)
+			if *intPtr != 0 {
+				out = msgp.AppendString(in, key)
+				out = msgp.AppendInt64(out, *intPtr)
 				return out, true
 			}
 			return in, false
 		},
 		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			var val bool
-			val, out, err = msgp.ReadBoolBytes(in)
-			if err == nil {
-				p.MetaRefineryShutdownSend.Set(val)
-			}
-			return out, err
-		},
-	},
-	{
-		key:          MetaRefinerySampleKey,
-		expectedType: FieldTypeString,
-		get: func(p *Payload) (value any, ok bool) {
-			if p.MetaRefinerySampleKey != "" {
-				return p.MetaRefinerySampleKey, true
-			}
-			return nil, false
-		},
-		set: func(p *Payload, value any) {
-			if v, ok := value.(string); ok {
-				p.MetaRefinerySampleKey = v
-			}
-		},
-		exist: func(p *Payload) bool {
-			return p.MetaRefinerySampleKey != ""
-		},
-		appendMsgp: func(p *Payload, in []byte) (out []byte, ok bool) {
-			if p.MetaRefinerySampleKey != "" {
-				out = msgp.AppendString(in, MetaRefinerySampleKey)
-				out = msgp.AppendString(out, p.MetaRefinerySampleKey)
-				return out, true
-			}
-			return in, false
-		},
-		unmarshalMsgp: func(p *Payload, in []byte) (out []byte, err error) {
-			p.MetaRefinerySampleKey, out, err = msgp.ReadStringBytes(in)
+			intPtr := ptr(p)
+			*intPtr, out, err = msgp.ReadInt64Bytes(in)
 			return out, err
 		},
-	},
-}
-
-// isMetadataField checks if a given key is a metadata field
-func isMetadataField(key string) bool {
-	for _, field := range metadataFields {
-		if field.key == key {
-			return true
-		}
 	}
-	return false
 }
 
 type nullableBool struct {
@@ -860,31 +310,23 @@ func (p *Payload) extractMetadataFromBytes(data []byte) (int, error) {
 		// Optimization: only check metadata fields if key starts with "meta."
 		if bytes.HasPrefix(keyBytes, []byte("meta.")) {
 			// Try to handle as a metadata field
-			for _, field := range metadataFields {
-				// Skip fields that don't match the expected type
+			if field, ok := metadataFields[string(keyBytes)]; ok {
+				// Check if field matches the expected type
+				var typeIsCorrect bool
 				switch field.expectedType {
 				case FieldTypeString:
-					if valueType != msgp.StrType {
-						continue
-					}
+					typeIsCorrect = valueType == msgp.StrType || valueType == msgp.BinType
 				case FieldTypeBool:
-					if valueType != msgp.BoolType {
-						continue
-					}
+					typeIsCorrect = valueType == msgp.BoolType
 				case FieldTypeInt64:
-					if valueType != msgp.IntType && valueType != msgp.UintType {
-						continue
-					}
+					typeIsCorrect = valueType == msgp.IntType || valueType == msgp.UintType
 				}
-
-				if bytes.Equal(keyBytes, []byte(field.key)) {
-					newRemaining, err := field.unmarshalMsgp(p, remaining)
+				if typeIsCorrect {
+					remaining, err = field.unmarshalMsgp(p, remaining)
 					if err != nil {
 						return len(data) - len(remaining), fmt.Errorf("failed to read value for key %s: %w", string(keyBytes), err)
 					}
-					remaining = newRemaining
 					handled = true
-					break
 				}
 			}
 		}
@@ -949,24 +391,21 @@ func (p *Payload) ExtractMetadata() error {
 		for key, value := range p.memoizedFields {
 			// Try metadata fields first
 			handled := false
-			for _, field := range metadataFields {
-				if field.key == key {
-					if field.expectedType == FieldTypeInt64 {
-						switch t := value.(type) {
-						case float64:
-							// JSON unmarshal will generally turn ints into floats.
-							field.set(p, int64(t))
-						case int:
-							field.set(p, int64(t))
-						default:
-							field.set(p, t)
-						}
-					} else {
-						field.set(p, value)
+			if field, ok := metadataFields[key]; ok {
+				if field.expectedType == FieldTypeInt64 {
+					switch t := value.(type) {
+					case float64:
+						// JSON unmarshal will generally turn ints into floats.
+						field.set(p, int64(t))
+					case int:
+						field.set(p, int64(t))
+					default:
+						field.set(p, t)
 					}
-					handled = true
-					break
+				} else {
+					field.set(p, value)
 				}
+				handled = true
 			}
 
 			// If not handled as metadata, check for trace/parent ID fields
@@ -1014,16 +453,16 @@ func NewPayload(config config.Config, data map[string]any) Payload {
 	return p
 }
 
-// UnmarshalMsgpack implements msgpack.Unmarshaler, but doesn't unmarshal. Instead it
-// keep a reference to serialized data.
+// UnmarshalMsgpack implements msgpack.Unmarshaler, but doesn't unmarshal.
+// Instead it keeps a copy of the serialized data.
 func (p *Payload) UnmarshalMsgpack(data []byte) error {
-	p.msgpMap = MsgpPayloadMap{rawData: data}
+	p.msgpMap = MsgpPayloadMap{rawData: slices.Clone(data)}
 	p.ExtractMetadata()
 	return nil
 }
 
 // UnmarshalMsg implements msgp.Unmarshaler, similar to above but expects to be
-// part of a larger message.
+// part of a larger message. Makes a local copy of the bytes it's hanging onto.
 func (p *Payload) UnmarshalMsg(bts []byte) (o []byte, err error) {
 	// Extract metadata and get consumed bytes
 	consumed, err := p.extractMetadataFromBytes(bts)
@@ -1032,7 +471,7 @@ func (p *Payload) UnmarshalMsg(bts []byte) (o []byte, err error) {
 	}
 
 	// Store the raw data
-	ourData := bts[:consumed]
+	ourData := slices.Clone(bts[:consumed])
 	p.msgpMap = MsgpPayloadMap{rawData: ourData}
 
 	// Return remainder
@@ -1092,7 +531,9 @@ func (p *Payload) MemoizeFields(keys ...string) {
 				break
 			}
 			key := string(keyBytes)
-			p.memoizedFields[key] = value
+
+			// Use Set here so we'll prefer metadata fields where appropriate.
+			p.Set(key, value)
 			keysFound++
 		}
 	}
@@ -1106,8 +547,8 @@ func (p *Payload) MemoizeFields(keys ...string) {
 
 func (p *Payload) Exists(key string) bool {
 	// if the key is a metadata field, check the dedicated field
-	for _, field := range metadataFields {
-		if field.key == key {
+	if strings.HasPrefix(key, "meta.") {
+		if field, ok := metadataFields[key]; ok {
 			return field.exist(p)
 		}
 	}
@@ -1149,13 +590,9 @@ func (p *Payload) Exists(key string) bool {
 func (p *Payload) Get(key string) any {
 	// Check if this is a metadata field and return from dedicated field
 	if strings.HasPrefix(key, "meta.") {
-		for _, field := range metadataFields {
-			if field.key == key {
-				if value, ok := field.get(p); ok {
-					return value
-				}
-				break
-			}
+		if field, ok := metadataFields[key]; ok {
+			value, _ := field.get(p)
+			return value
 		}
 	}
 
@@ -1196,11 +633,9 @@ func (p *Payload) Get(key string) any {
 
 func (p *Payload) Set(key string, value any) {
 	// Check if this is a metadata field and update dedicated field
-	for _, field := range metadataFields {
-		if field.key == key {
-			field.set(p, value)
-			break
-		}
+	if field, ok := metadataFields[key]; ok {
+		field.set(p, value)
+		return
 	}
 
 	if p.memoizedFields == nil {
@@ -1220,16 +655,17 @@ func (p *Payload) IsEmpty() bool {
 func (p *Payload) All() iter.Seq2[string, any] {
 	return func(yield func(string, any) bool) {
 		// First yield metadata fields with non-default values
-		for _, field := range metadataFields {
-			if field.key == MetaTraceID {
+		for key, field := range metadataFields {
+			if key == MetaTraceID {
 				continue
 			}
-			if field.key == MetaRefineryRoot && !p.MetaRefineryRoot.Value {
+			if key == MetaRefineryRoot && !p.MetaRefineryRoot.Value {
 				// Skip the root field if it's false, as it doesn't need to be yielded
 				continue
 			}
+
 			if value, ok := field.get(p); ok {
-				if !yield(field.key, value) {
+				if !yield(key, value) {
 					return
 				}
 			}
@@ -1237,6 +673,10 @@ func (p *Payload) All() iter.Seq2[string, any] {
 
 		// Then yield memoized fields
 		for key, value := range p.memoizedFields {
+			// Skip metadata fields.
+			if _, ok := metadataFields[key]; ok {
+				continue
+			}
 			if !yield(key, value) {
 				return
 			}
@@ -1265,7 +705,7 @@ func (p *Payload) All() iter.Seq2[string, any] {
 			}
 
 			// Skip metadata fields as they're already yielded
-			if isMetadataField(key) {
+			if _, ok := metadataFields[key]; ok {
 				continue
 			}
 
@@ -1361,6 +801,11 @@ func (p Payload) MarshalMsg(buf []byte) ([]byte, error) {
 
 	// Serialize regular memoized fields
 	for key, value := range p.memoizedFields {
+		// Skip metadata fields as they're serialized separately
+		if _, ok := metadataFields[key]; ok {
+			continue
+		}
+
 		buf = msgp.AppendString(buf, key)
 		var err error
 		buf, err = msgp.AppendIntf(buf, value)
@@ -1386,7 +831,7 @@ func (p Payload) MarshalMsg(buf []byte) ([]byte, error) {
 			}
 
 			// Skip metadata fields as they're serialized separately
-			if isMetadataField(keyStr) {
+			if _, ok := metadataFields[keyStr]; ok {
 				continue
 			}
 


Test patch:
diff --git a/route/route_test.go b/route/route_test.go
index d1f0afb..77a6445 100644
--- a/route/route_test.go
+++ b/route/route_test.go
@@ -56,7 +56,7 @@ func TestDecompression(t *testing.T) {
 		Body:   io.NopCloser(pReader),
 		Header: http.Header{},
 	}
-	reader, err := router.getMaybeCompressedBody(req)
+	reader, err := router.readAndCloseMaybeCompressedBody(req)
 	if err != nil {
 		t.Errorf("unexpected err: %s", err.Error())
 	}
@@ -79,7 +79,7 @@ func TestDecompression(t *testing.T) {
 
 	req.Body = io.NopCloser(buf)
 	req.Header.Set("Content-Encoding", "gzip")
-	reader, err = router.getMaybeCompressedBody(req)
+	reader, err = router.readAndCloseMaybeCompressedBody(req)
 	if err != nil {
 		t.Errorf("unexpected err: %s", err.Error())
 	}
@@ -105,7 +105,7 @@ func TestDecompression(t *testing.T) {
 
 	req.Body = io.NopCloser(buf)
 	req.Header.Set("Content-Encoding", "zstd")
-	reader, err = router.getMaybeCompressedBody(req)
+	reader, err = router.readAndCloseMaybeCompressedBody(req)
 	if err != nil {
 		t.Errorf("unexpected err: %s", err.Error())
 	}
@@ -1212,7 +1212,7 @@ func createBatchEventsWithLargeAttributes(numEvents int, cfg config.Config) *bat
 func BenchmarkRouterBatch(b *testing.B) {
 	router := newBatchRouter(b)
 	batchEvents := createBatchEventsWithLargeAttributes(3, router.Config)
-	batchMsgpack, err := msgpack.Marshal(batchEvents)
+	batchMsgpack, err := msgpack.Marshal(batchEvents.events)
 	require.NoError(b, err)
 
 	mockCollector := router.Collector.(*collect.MockCollector)
diff --git a/transmit/direct_transmit_test.go b/transmit/direct_transmit_test.go
index 3c91671..dfdab66 100644
--- a/transmit/direct_transmit_test.go
+++ b/transmit/direct_transmit_test.go
@@ -62,20 +62,19 @@ func newTestDirectAPIServer(t testing.TB, maxBatchSize int) *testDirectAPIServer
 	return server
 }
 
-// Not suitable for use in benchmarks
 func readHTTPBody(t testing.TB, r *http.Request) []byte {
 	t.Helper()
 
-	var reader io.Reader
+	body, err := io.ReadAll(r.Body)
+	assert.NoError(t, err)
+	assert.Len(t, body, int(r.ContentLength))
+
 	if r.Header.Get("Content-Encoding") == "zstd" {
-		zr, err := zstd.NewReader(r.Body)
+		zr, err := zstd.NewReader(bytes.NewReader(body))
+		assert.NoError(t, err)
+		body, err = io.ReadAll(zr)
 		assert.NoError(t, err)
-		reader = zr
-	} else {
-		reader = r.Body
 	}
-	body, err := io.ReadAll(reader)
-	assert.NoError(t, err)
 	return body
 }
 
diff --git a/types/payload_test.go b/types/payload_test.go
index d93a742..caefe6f 100644
--- a/types/payload_test.go
+++ b/types/payload_test.go
@@ -10,13 +10,12 @@ import (
 	"github.com/honeycombio/refinery/config"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/tinylib/msgp/msgp"
 	"github.com/vmihailenco/msgpack/v5"
 )
 
 func TestMetadataFieldsHaveMetaPrefix(t *testing.T) {
-	for _, field := range metadataFields {
-		assert.True(t, strings.HasPrefix(field.key, "meta."))
+	for key := range metadataFields {
+		assert.True(t, strings.HasPrefix(key, "meta."))
 	}
 }
 
@@ -32,6 +31,9 @@ func TestPayload(t *testing.T) {
 		"meta.refinery.root":     true,
 		"meta.refinery.min_span": false,
 		"meta.annotation_type":   "span_event",
+
+		// This is the wrong type, it should just be ignored and dropped from the data.
+		"meta.refinery.send_by": "never",
 	}
 	mockCfg := &config.MockConfig{
 		TraceIdFieldNames: []string{"trace.trace_id"},
@@ -61,6 +63,9 @@ func TestPayload(t *testing.T) {
 		assert.True(t, ph.Exists("meta.annotation_type"))
 		assert.Equal(t, "span_event", ph.Get("meta.annotation_type"))
 
+		assert.Nil(t, ph.Get("meta.refinery.send_by"))
+		assert.False(t, ph.Exists("meta.refinery.send_by"))
+
 		// Test dedicated fields - should be populated from initial data or unmarshal
 		assert.Equal(t, "test-trace-123", ph.MetaTraceID, "MetaTraceID should be populated")
 		assert.True(t, ph.MetaRefineryRoot.HasValue, "MetaRefineryRoot should be set")
@@ -68,6 +73,7 @@ func TestPayload(t *testing.T) {
 		assert.True(t, ph.MetaRefineryMinSpan.HasValue, "MetaRefineryMinSpan should be set")
 		assert.Equal(t, false, ph.MetaRefineryMinSpan.Value, "MetaRefineryMinSpan should be populated")
 		assert.Equal(t, "span_event", ph.MetaAnnotationType, "MetaAnnotationType should be populated")
+		assert.Equal(t, int64(0), ph.MetaRefinerySendBy)
 
 		ph.Set("key5", "newvalue")
 		assert.True(t, ph.Exists("key5"))
@@ -96,6 +102,7 @@ func TestPayload(t *testing.T) {
 		expected["key5"] = "newvalue"
 		expected["meta.refinery.span_data_size"] = int64(1234)
 		expected["meta.refinery.incoming_user_agent"] = "test-agent/1.0"
+		delete(expected, "meta.refinery.send_by")
 		found := maps.Collect(ph.All())
 		assert.Equal(t, expected, found)
 
@@ -116,7 +123,6 @@ func TestPayload(t *testing.T) {
 		// round-tripping through JSON turns our ints into floats
 		expectedFromJSON := maps.Collect(ph.All())
 		expectedFromJSON["key2"] = 42.0
-		expectedFromJSON["meta.refinery.span_data_size"] = 1234.0
 		assert.EqualValues(t, expectedFromJSON, maps.Collect(fromJSON.All()))
 	}
 
@@ -234,55 +240,54 @@ func TestPayloadExtractMetadataError(t *testing.T) {
 	})
 }
 
-func TestPayloadMetaAnnotationType(t *testing.T) {
-	t.Run("handles string values", func(t *testing.T) {
-		ph := NewPayload(&config.MockConfig{}, map[string]any{
-			"meta.annotation_type": "span_event",
-		})
-		err := ph.ExtractMetadata()
-		require.NoError(t, err)
-
-		assert.Equal(t, "span_event", ph.MetaAnnotationType)
-		assert.Equal(t, "span_event", ph.Get("meta.annotation_type"))
-	})
-
-	t.Run("marshaling preserves type", func(t *testing.T) {
-		// Test string marshaling
-		ph1 := NewPayload(&config.MockConfig{}, nil)
-		ph1.MetaAnnotationType = "span_event"
-
-		msgpData, err := ph1.MarshalMsg(nil)
-		require.NoError(t, err)
-
-		ph2 := NewPayload(&config.MockConfig{}, nil)
-		_, err = ph2.UnmarshalMsg(msgpData)
-		require.NoError(t, err)
-
-		assert.Equal(t, "span_event", ph2.MetaAnnotationType)
-	})
-}
-
 func TestPayloadUnmarshalMsg(t *testing.T) {
 	t.Run("extracts metadata during unmarshal", func(t *testing.T) {
-		// Create test data with metadata fields
+		// Create test data with all metadata fields
 		data := map[string]any{
-			"regular_field":        "value1",
+			// Regular fields
+			"regular_field": "value1",
+			"another_field": 42,
+
+			// Core metadata fields
 			"meta.signal_type":     "trace",
 			"meta.annotation_type": "span_event",
-			"meta.refinery.root":   true,
-			"meta.refinery.probe":  false,
-			"trace.trace_id":       "custom-trace-456", // Custom trace ID field
-			"span.parent_id":       "",                 // Empty parent ID should make it root
+
+			// Refinery boolean fields
+			"meta.refinery.probe":         true,
+			"meta.refinery.min_span":      true,
+			"meta.refinery.expired_trace": false,
+			"meta.refinery.shutdown_send": true,
+			"meta.stressed":               true,
+
+			// Refinery string fields
+			"meta.refinery.incoming_user_agent": "test-agent/1.0",
+			"meta.refinery.forwarded":           "192.168.1.1",
+			"meta.refinery.local_hostname":      "test-host",
+			"meta.refinery.reason":              "deterministic",
+			"meta.refinery.send_reason":         "trace_timeout",
+			"meta.refinery.sample_key":          "sample-key-123",
+
+			// Refinery int64 fields
+			"meta.refinery.send_by":              int64(1234567890),
+			"meta.refinery.span_data_size":       int64(2048),
+			"meta.span_event_count":              int64(5),
+			"meta.span_link_count":               int64(3),
+			"meta.span_count":                    int64(10),
+			"meta.event_count":                   int64(15),
+			"meta.refinery.original_sample_rate": int64(100),
+
+			// Custom trace/parent fields
+			"trace.trace_id": "custom-trace-456", // Custom trace ID field
+			"span.parent_id": "",                 // Empty parent ID should make it root
 		}
 
-		// Marshal to msgpack
 		msgpData, err := msgpack.Marshal(data)
 		require.NoError(t, err)
 
 		// Create payload and unmarshal with metadata extraction
 		p := Payload{
 			config: &config.MockConfig{
-				TraceIdFieldNames:  []string{"trace.trace_id"},
+				TraceIdFieldNames:  []string{"wrong", "trace.trace_id"},
 				ParentIdFieldNames: []string{"span.parent_id"},
 			},
 		}
@@ -290,17 +295,45 @@ func TestPayloadUnmarshalMsg(t *testing.T) {
 		require.NoError(t, err)
 		assert.Empty(t, remainder)
 
-		// Verify metadata was extracted
+		// Verify core metadata was extracted
 		assert.Equal(t, "trace", p.MetaSignalType)
-		assert.Equal(t, "custom-trace-456", p.MetaTraceID) // Should use custom field
+		assert.Equal(t, "custom-trace-456", p.MetaTraceID) // Should use custom field over meta.trace_id
 		assert.Equal(t, "span_event", p.MetaAnnotationType)
-		assert.True(t, p.MetaRefineryRoot.HasValue)
-		assert.True(t, p.MetaRefineryRoot.Value)
+
+		// Verify boolean metadata fields
 		assert.True(t, p.MetaRefineryProbe.HasValue)
-		assert.False(t, p.MetaRefineryProbe.Value)
+		assert.True(t, p.MetaRefineryProbe.Value)
+		assert.True(t, p.MetaRefineryRoot.HasValue)
+		assert.True(t, p.MetaRefineryRoot.Value) // Should be true because empty parent ID
+		assert.True(t, p.MetaRefineryMinSpan.HasValue)
+		assert.True(t, p.MetaRefineryMinSpan.Value)
+		assert.True(t, p.MetaRefineryExpiredTrace.HasValue)
+		assert.False(t, p.MetaRefineryExpiredTrace.Value)
+		assert.True(t, p.MetaRefineryShutdownSend.HasValue)
+		assert.True(t, p.MetaRefineryShutdownSend.Value)
+		assert.True(t, p.MetaStressed.HasValue)
+		assert.True(t, p.MetaStressed.Value)
+
+		// Verify string metadata fields
+		assert.Equal(t, "test-agent/1.0", p.MetaRefineryIncomingUserAgent)
+		assert.Equal(t, "192.168.1.1", p.MetaRefineryForwarded)
+		assert.Equal(t, "test-host", p.MetaRefineryLocalHostname)
+		assert.Equal(t, "deterministic", p.MetaRefineryReason)
+		assert.Equal(t, "trace_timeout", p.MetaRefinerySendReason)
+		assert.Equal(t, "sample-key-123", p.MetaRefinerySampleKey)
+
+		// Verify int64 metadata fields
+		assert.Equal(t, int64(1234567890), p.MetaRefinerySendBy)
+		assert.Equal(t, int64(2048), p.MetaRefinerySpanDataSize)
+		assert.Equal(t, int64(5), p.MetaSpanEventCount)
+		assert.Equal(t, int64(3), p.MetaSpanLinkCount)
+		assert.Equal(t, int64(10), p.MetaSpanCount)
+		assert.Equal(t, int64(15), p.MetaEventCount)
+		assert.Equal(t, int64(100), p.MetaRefineryOriginalSampleRate)
 
 		// Verify regular fields are still accessible
 		assert.Equal(t, "value1", p.Get("regular_field"))
+		assert.Equal(t, int64(42), p.Get("another_field")) // should preserve int type
 	})
 
 	t.Run("handles remainder correctly", func(t *testing.T) {
@@ -334,6 +367,7 @@ func TestPayloadUnmarshalMsg(t *testing.T) {
 		assert.Empty(t, remainder2)
 	})
 }
+
 func TestPayloadGetSetExistMetadataSync(t *testing.T) {
 	t.Run("Set updates metadata fields", func(t *testing.T) {
 		ph := NewPayload(&config.MockConfig{}, nil)
@@ -423,79 +457,6 @@ func TestPayloadGetSetExistMetadataSync(t *testing.T) {
 	})
 }
 
-func TestUnmarshalMsgWithMetadata(t *testing.T) {
-	// Create test data with metadata fields
-	data := map[string]interface{}{
-		"trace.trace_id":                    "test-trace-123",
-		"trace.parent_id":                   "parent-456",
-		"meta.signal_type":                  "span",
-		"meta.annotation_type":              "span_event",
-		"meta.refinery.probe":               true,
-		"meta.refinery.root":                false,
-		"meta.refinery.incoming_user_agent": "test-agent",
-		"meta.refinery.send_by":             int64(1234567890),
-		"meta.refinery.span_data_size":      int64(1024),
-		"meta.refinery.min_span":            true,
-		"meta.refinery.forwarded":           "192.168.1.1",
-		"meta.refinery.expired_trace":       false,
-		"regular_field":                     "value",
-		"another_field":                     42,
-	}
-
-	// Marshal the data to msgpack
-	var buf []byte
-	buf = msgp.AppendMapHeader(buf, uint32(len(data)))
-	for k, v := range data {
-		buf = msgp.AppendString(buf, k)
-		switch val := v.(type) {
-		case string:
-			buf = msgp.AppendString(buf, val)
-		case int64:
-			buf = msgp.AppendInt64(buf, val)
-		case int:
-			buf = msgp.AppendInt(buf, val)
-		case bool:
-			buf = msgp.AppendBool(buf, val)
-		}
-	}
-
-	// Test the optimized unmarshal
-	p := Payload{
-		config: &config.MockConfig{
-			TraceIdFieldNames:  []string{"trace.trace_id", "traceId"},
-			ParentIdFieldNames: []string{"span.parent_id", "parentId"},
-		},
-	}
-
-	remaining, err := p.UnmarshalMsg(buf)
-	require.NoError(t, err)
-	assert.Empty(t, remaining)
-
-	// Verify metadata was extracted correctly
-	assert.Equal(t, "test-trace-123", p.MetaTraceID)
-	assert.Equal(t, "span", p.MetaSignalType)
-	assert.Equal(t, "span_event", p.MetaAnnotationType)
-	assert.True(t, p.MetaRefineryProbe.HasValue)
-	assert.True(t, p.MetaRefineryProbe.Value)
-	assert.True(t, p.MetaRefineryRoot.HasValue)
-	assert.False(t, p.MetaRefineryRoot.Value)
-	assert.Equal(t, "test-agent", p.MetaRefineryIncomingUserAgent)
-	assert.Equal(t, int64(1234567890), p.MetaRefinerySendBy)
-	assert.Equal(t, int64(1024), p.MetaRefinerySpanDataSize)
-	assert.True(t, p.MetaRefineryMinSpan.HasValue)
-	assert.True(t, p.MetaRefineryMinSpan.Value)
-	assert.Equal(t, "192.168.1.1", p.MetaRefineryForwarded)
-	assert.True(t, p.MetaRefineryExpiredTrace.HasValue)
-	assert.False(t, p.MetaRefineryExpiredTrace.Value)
-
-	// Verify HasExtractedMetadata returns true
-	assert.True(t, p.hasExtractedMetadata)
-
-	// Verify we can still access regular fields
-	assert.Equal(t, "value", p.Get("regular_field"))
-	assert.Equal(t, int64(42), p.Get("another_field"))
-}
-
 func BenchmarkPayload(b *testing.B) {
 	// Create test data with many fields
 	var keys []string