reliar-store-postgres 0.9.0

//! N10 (`docs/architecture/outbox-enqueue-contract.md` §8): exactly one `reliar.outbox.enqueue`
//! span per enqueued envelope, carrying `message.id`/`message.type`; no batch span (the crate has
//! no batch method); and no payload byte or header value in any span field or event (recording-
//! subscriber test, replacing the deleted `reliar-outbox` test of the same shape once
//! `OutboxPublisher` — and its `publish`/`publish_batch` no-span guarantee — was withdrawn,
//! RELIAR-61).

use crate::common;

use std::fmt::Write as _;
use std::sync::{Arc, Mutex};

use crate::common::OrderCreated;
use reliar_core::{ContentType, Envelope, Message, Serializer};
use reliar_outbox::OutboxEnqueue;
use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
use tracing::field::{Field, Visit};
use tracing_subscriber::layer::{Context, SubscriberExt};

const SECRET_PAYLOAD_MARKER: &str = "sk_live_RELIAR_PAYLOAD_MUST_NEVER_APPEAR_IN_A_LOG";
const SECRET_HEADER_VALUE: &str = "RELIAR_HEADER_VALUE_MUST_NEVER_APPEAR_IN_A_LOG";

/// A body with a real string field — `common::OrderCreated`'s `u64` cannot carry a secret marker
/// through `JsonSerializer`, and this test needs the marker to actually reach the serialized
/// `payload` column, not just a header, to prove the span never quotes it.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct SecretPayload {
    secret: String,
}

impl reliar_core::Message for SecretPayload {
    const TYPE: &'static str = "secrets.logged";
    const VERSION: u16 = 1;
}

/// Renders every recorded field as `name=value ` into the caller's line buffer — shared by
/// span-open and event recording so both land in the same transcript format.
struct TranscriptVisitor<'a>(&'a mut String);

impl Visit for TranscriptVisitor<'_> {
    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
        let _ = write!(self.0, "{}={value:?} ", field.name());
    }
}

#[derive(Default, Clone)]
struct Transcript(Arc<Mutex<String>>);

impl Transcript {
    fn text(&self) -> String {
        self.0.lock().unwrap().clone()
    }
}

/// A recording `tracing_subscriber::Layer` (the same pattern as `outbox_schema_verification.rs`'s
/// `Recorder`, extended to spans): every span open and every event becomes one line, `name{fields}`
/// or `name fields`, in the shared [`Transcript`] — a plain grep target, not a structured API,
/// since the point is "nothing sensitive appears anywhere", not "field X equals Y".
struct Recorder(Transcript);

impl<S> tracing_subscriber::Layer<S> for Recorder
where
    S: tracing::Subscriber + for<'lookup> tracing_subscriber::registry::LookupSpan<'lookup>,
{
    fn on_new_span(
        &self,
        attrs: &tracing::span::Attributes<'_>,
        _id: &tracing::span::Id,
        _ctx: Context<'_, S>,
    ) {
        let mut line = format!("SPAN_OPEN {}{{ ", attrs.metadata().name());

        attrs.record(&mut TranscriptVisitor(&mut line));
        line.push_str("}\n");

        self.0.0.lock().unwrap().push_str(&line);
    }

    fn on_event(&self, event: &tracing::Event<'_>, ctx: Context<'_, S>) {
        // Names the event's *current* span (if any) so a caller can assert an event landed
        // inside a specific span, not merely that it fired somewhere in the transcript.
        let current_span = ctx.lookup_current().map_or("-", |s| s.name());
        let mut line = format!("EVENT[{current_span}] {} ", event.metadata().name());

        event.record(&mut TranscriptVisitor(&mut line));
        line.push('\n');

        self.0.0.lock().unwrap().push_str(&line);
    }
}

/// Async-friendly: `set_default` returns a guard that stays active across `.await` points on the
/// current OS thread, matching a `#[tokio::test]`'s single thread. Goes through
/// [`common::install_recording_subscriber`] so the process-wide callsite
/// interest cache is rebuilt right after installing — otherwise a concurrent trial's callsite
/// hit can cache this recorder's spans/events as `never` before it ever gets a chance to see them.
fn install_recorder() -> (Transcript, tracing::subscriber::DefaultGuard) {
    let transcript = Transcript::default();
    let subscriber = tracing_subscriber::registry().with(Recorder(transcript.clone()));
    let guard = common::install_recording_subscriber(subscriber);

    (transcript, guard)
}

async fn enqueue_emits_one_span_per_envelope_with_no_leak() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone());

    let (transcript, _guard) = install_recorder();

    let first = Envelope::builder(SecretPayload {
        secret: SECRET_PAYLOAD_MARKER.to_string(),
    })
    .header("x-secret", SECRET_HEADER_VALUE)
    .expect("a non-reserved header key is accepted")
    .build();
    let second = Envelope::builder(OrderCreated { order_id: 2 }).build();
    let (first_id, second_id) = (first.id, second.id);

    let mut tx = pool.begin().await.unwrap();

    store.enqueue_envelope(&mut tx, first).await.unwrap();
    store.enqueue_envelope(&mut tx, second).await.unwrap();
    // RELIAR-63 N10: the bare spelling reaches the same `enqueue_envelope` impl, so it must fire
    // the same span exactly once too — not just the envelope spelling exercised above.
    let third_id = store
        .enqueue(&mut tx, OrderCreated { order_id: 3 })
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let text = transcript.text();

    // Not just "the string `reliar.outbox.enqueue` appears" — grep the span-open marker itself so
    // a stray substring match (e.g. inside some other span's name) cannot inflate the count.
    let enqueue_span_count = text.matches("SPAN_OPEN reliar.outbox.enqueue{").count();
    assert_eq!(
        enqueue_span_count, 3,
        "expected exactly one `reliar.outbox.enqueue` span per envelope, for either spelling:\n{text}"
    );
    assert!(
        text.contains(&format!("message.id={third_id}")),
        "message.id must be recorded on the bare-spelling span:\n{text}"
    );
    assert!(
        !text.contains("reliar.outbox.enqueue_batch"),
        "the crate has no batch method — no batch span may appear:\n{text}"
    );
    assert!(
        text.contains(&format!("message.id={first_id}")),
        "message.id must be recorded on the first envelope's span:\n{text}"
    );
    assert!(
        text.contains(&format!("message.id={second_id}")),
        "message.id must be recorded on the second envelope's span:\n{text}"
    );
    assert!(
        text.contains("message.type=secrets.logged")
            && text.contains("message.type=orders.created"),
        "message.type must be recorded on each enqueue span:\n{text}"
    );
    assert!(
        !text.contains(SECRET_PAYLOAD_MARKER),
        "payload leaked into a span field or event:\n{text}"
    );
    assert!(
        !text.contains(SECRET_HEADER_VALUE),
        "header value leaked into a span field or event:\n{text}"
    );
}

/// R63-N.M8: a serializer whose `serialize` emits its own `tracing::debug!` event — proves the
/// event lands *inside* `reliar.outbox.enqueue`, not before it, now that serialization runs
/// under an entered span guard rather than only inside the later `.instrument`ed future.
#[derive(Clone, Debug, Default)]
struct DebugEmittingSerializer;

impl Serializer for DebugEmittingSerializer {
    type Error = std::convert::Infallible;

    fn content_type(&self) -> &ContentType {
        static CONTENT_TYPE: std::sync::LazyLock<ContentType> =
            std::sync::LazyLock::new(|| ContentType::parse("application/json").unwrap());

        &CONTENT_TYPE
    }

    fn serialize<T: Message>(&self, body: &T) -> Result<bytes::Bytes, Self::Error> {
        tracing::debug!("serializing a body");

        Ok(serde_json::to_vec(body).map(bytes::Bytes::from).unwrap())
    }

    fn deserialize<T: Message>(&self, bytes: &[u8]) -> Result<T, Self::Error> {
        Ok(serde_json::from_slice(bytes).unwrap())
    }
}

async fn serializer_events_land_inside_the_enqueue_span() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::with_serializer(
        pool.clone(),
        PostgresOutboxSettings::default(),
        DebugEmittingSerializer,
    );

    let (transcript, _guard) = install_recorder();
    let mut tx = pool.begin().await.unwrap();

    store
        .enqueue(&mut tx, OrderCreated { order_id: 1 })
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let text = transcript.text();
    // The exact event line, e.g. `EVENT[reliar.outbox.enqueue] event <file>:<line>
    // message="serializing a body"` — match the two anchors, not the file:line in between.
    assert!(
        text.lines()
            .any(|line| line.starts_with("EVENT[reliar.outbox.enqueue]")
                && line.contains("serializing a body")),
        "the serializer's own event must be recorded inside the enqueue span:\n{text}"
    );
}

/// Both trials in this file install a thread-local recording subscriber (`install_recorder`) —
/// they must run in `main.rs`'s serialised recorder phase, never in the parallel batch
/// (RELIAR-66; see `common::install_recording_subscriber`'s doc for why).
pub(crate) fn recorder_trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "outbox_enqueue_spans::enqueue_emits_one_span_per_envelope_with_no_leak",
            move || {
                rt.block_on(enqueue_emits_one_span_per_envelope_with_no_leak());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue_spans::serializer_events_land_inside_the_enqueue_span",
            move || {
                rt.block_on(serializer_events_land_inside_the_enqueue_span());
                Ok(())
            },
        ),
    ]
}