Set Kafka timestamps and expose record headers

[jakub-petrylak-onerail]

Summary

• Set Envelope.SentAt from the native Kafka record timestamp when available.

• Copy Kafka record headers into Wolverine envelope headers without overwriting existing values.

Why

Using Kafka’s record timestamp gives consumers an accurate transport-level reference point for measuring end-to-end message latency. It also distinguishes message production or broker-append time from business-event and consumer-processing timestamps.

Exposing Kafka headers through the Wolverine envelope makes broker metadata readily available to handlers and middleware without requiring direct access to Kafka-specific APIs.

[jeremydmiller]

Thanks for this — the direction is right and the gap is real. For a raw-JSON listener there genuinely is no Wolverine metadata on the wire, so stamping SentAt from the broker's record timestamp and surfacing the record headers is exactly the missing capability. I also like that you scoped it to JsonOnlyMapper and left the Wolverine-to-Wolverine path alone, where sent-at and the headers already ride in the envelope protocol. The unit tests are well-chosen.

Two things before I can merge, one of them a genuine security issue.

1. Blocking: the header copy must exclude Wolverine's reserved keys

The copy loop is unfiltered:

foreach (var header in incoming.Headers)
{
    if (TryReadHeader(incoming, header.Key, out var headerValue))
    {
        envelope.Headers.TryAdd(header.Key, headerValue);
    }
}

Kafka header keys are arbitrary strings chosen by the producer, and on a raw-JSON topic that producer is by definition not Wolverine and not necessarily trusted. EnvelopeConstants reserves a set of those names — id, message-type, saga-id, tenant-id, correlation-id, destination, reply-uri, and others.

Right now that looks inert, because JsonOnlyMapper doesn't promote headers into typed properties. But it does not stay inert. In EnvelopeSerializer.writeHeaders (src/Wolverine/Runtime/Serialization/EnvelopeSerializer.cs:287), the typed properties are written first and then every env.Headers entry is appended verbatim, with no reserved-key filter:

writer.WriteProp(ref count, EnvelopeConstants.TenantIdKey, env.TenantId);
...
foreach (var pair in env.Headers)   // <-- unfiltered, and it comes last
{
    count++;
    writer.Write(pair.Key);
    writer.Write(pair.Value);
}

and the reader switches reserved keys straight back into typed properties (env.TenantId = value, env.SagaId = value, env.Id = ...), processing entries in order — so the appended copy wins.

So on a durable Kafka listener the sequence is:

1. An external producer puts tenant-id: acme on the topic.
2. This PR copies it into envelope.Headers["tenant-id"]. Still inert — envelope.TenantId is null.
3. The durable inbox persists the envelope. writeHeaders appends tenant-id = acme after the (null) typed prop.
4. On read back, case EnvelopeConstants.TenantIdKey: env.TenantId = value; fires.

A header supplied by an untrusted external producer has become the envelope's TenantId. saga-id gets you someone else's saga state, and id rewrites Envelope.Id, which is the inbox's dedupe identity. That's a privilege boundary, not a style nit.

It's worth noting your own test picks tenant-id as its example custom header — which is the trap in miniature, and honestly why I want the guard rather than a doc note.

Fix: skip reserved keys when copying. Something like a static readonly FrozenSet<string> of the EnvelopeConstants key names, and if (Reserved.Contains(header.Key)) continue; in the loop. Please add a test that an incoming Kafka tenant-id / saga-id / id header does not land on the envelope, so this can't regress.

(If you'd rather have the reserved-key filter live in EnvelopeSerializer.writeHeaders instead, say so — that arguably hardens every transport at once and I'd take that as a separate PR. But this PR is what makes the path reachable, so it needs to be closed here either way.)

2. The integration timestamp test is vacuous

received.SentAt.ShouldBeGreaterThan(DateTimeOffset.UtcNow.Subtract(5.Minutes()));

Envelope.SentAt is initialized to DateTimeOffset.UtcNow at construction (src/Wolverine/Envelope.cs:248), so a freshly received envelope already satisfies this assertion with or without your change. It passes on main today. It's false confidence rather than coverage.

Your unit tests already prove the mapping properly (they null out SentAt to MinValue first and assert the exact value), so this one just needs teeth. The easy strengthening: you're already standing up a raw ProducerBuilder in copies_kafka_headers_onto_the_received_envelope — do the same here, produce with an explicit Timestamp well in the past (say CreateTime an hour ago), and assert the received SentAt equals it. That distinguishes "came from the record" from "came from the constructor", which is the whole point of the change.

Minor

• if (incoming.Headers is null) return; — the codebase braces its single-statement ifs; please match.
• Not a blocker, but the PR title says "Set Kafka timestamps" while the change only reads the incoming record timestamp (Confluent assigns the outgoing one at produce time). "Read" would describe it better.

Happy to take this once (1) is closed and (2) has teeth — and thank you for the tests, the unit coverage in particular is better than a lot of what comes through here.

[mysticmind]

@jakub-petrylak-onerail Can you please take a look at Jeremy's comments and address the same?

[jakub-petrylak-onerail]

Thanks for the review — I've addressed Jeremy's feedback:

1. Reserved header keys — the copy loop now skips EnvelopeSerializer.ReservedHeaderKeys, with unit and integration tests asserting that incoming tenant-id / saga-id / id headers do not land on the envelope.
2. Integration timestamp test — stamps_envelope_sent_at_from_the_kafka_record_timestamp now produces with an explicit CreateTime an hour in the past and asserts the exact SentAt, so it can't pass on the constructor default alone.

While working on (2), I think I also uncovered a pre-existing bug in ReceiveRawJson in KafkaListenerConfiguration.cs:

// before
return UseInterop((e, _) => new JsonOnlyMapper(e, options ?? new()));

// after
return UseInterop((_, e) => new JsonOnlyMapper(e, options ?? new()));

UseInterop has two relevant overloads on listeners:

• UseInterop(Action<TEndpoint, TConcreteMapper> configure) → calls customizeMapping on the default KafkaEnvelopeMapper
• UseInterop(Func<IWolverineRuntime, TEndpoint, TMapper> factory) → calls registerMapperFactory and actually wires up a custom mapper

With (e, _), the compiler resolves to the Action<...> overload (first param = KafkaTopic, which type-checks for JsonOnlyMapper). The new JsonOnlyMapper(...) return value is discarded, so JsonOnlyMapper is never registered — the endpoint keeps using the default KafkaEnvelopeMapper instead.

With (_, e), it resolves to the Func<...> factory overload (second param = KafkaTopic), which is what ReceiveRawJson intends.

I confirmed this with the integration test: with the old parameter order, stamps_envelope_sent_at_from_the_kafka_record_timestamp fails (SentAt stays at DateTimeOffset.MinValue / constructor default) because timestamp stamping lives in JsonOnlyMapper, not the default mapper. With the swap it passes.

Could you double-check that this reading is correct? If so, the fix is included in the latest push. (Side note: PublishRawJson on KafkaSubscriberConfiguration still uses (e, _), which may have the same overload-resolution issue on the subscriber side — happy to fix that in a follow-up if you want.)