Outbound Relay — Durable JetStream → Telegram sendMessage

Status: LIVE in production (pin 3baa6e4). Fully implemented with 23 passing tests, integration tests via wiremock.

Design Principles

Single responsibility: Consume from JetStream, send to Telegram, manage ack/nak/term. No trading logic, no other service’s DB.
Verbatim relay: Publisher owns message formatting and escaping. Connector never re-escapes.
Durability: At-least-once delivery via NATS pull consumer (message persists until ack).
Reactive rate limiting: Honor Telegram’s 429 retry_after without custom backoff algorithms.
Resilience: Survive transient network errors, reconnect on stream-end, never crash on Telegram API errors.

Message Flow

Publisher (cohort_algo, etc.)
    ↓ publish to telegram.outbound
NATS JetStream TELEGRAM_OUTBOUND stream (1d retention, 5m dedup)
    ↓ pull consumer tg_outbound (durable)
telegram_connector (async poll loop)
    ↓ classify response
Bot API sendMessage
    ↓
Telegram API

Wire Contract

Type: TelegramOutbound from pmv2-contracts

pub struct TelegramOutbound {
    pub msg_id: String,              // idempotency key; also = Nats-Msg-Id
    pub chat: Option<String>,        // null → use TELEGRAM_CHAT_ID env
    pub text: String,                // formatted body (publisher owns escaping)
    pub parse_mode: Option<String>,  // "HTML" | "MarkdownV2" | null
    pub kind: Option<String>,        // "info" | "alert" | "trade" (metadata)
}

Publisher: Any pmv2 service publishes a TelegramOutbound JSON to telegram.outbound with Nats-Msg-Id: <msg_id>.

Example: from cohort_algo:

{
  "msg_id": "cohort|1234|signal_id",
  "chat": null,
  "text": "<b>Signal</b>: BUY at 0.45",
  "parse_mode": "HTML",
  "kind": "trade"
}

Delivery Guarantees

At-Least-Once via Ack/Nak/Term

Each message is processed in one of three terminal states:

`Sent` → Ack (advance consumer offset)

Telegram API returned 200 OK with a message_id. Message is “sent” — ack the NATS consumer offset and move to next message.

`Transient` → Nak with backoff (retry)

Telegram returned a retryable error (429, 5xx, or network timeout). Negative-ack the message:

NATS resets the consumer offset, message re-delivered on next poll
Apply backoff: if 429 retry_after header present, use it (capped to ack_wait); else exponential backoff (30s default)
Try again

`Terminal` → Term + drop + alert

Telegram returned a non-retryable error (400 Bad Request, 403 Forbidden, 401 Unauthorized). The message is bad (malformed, invalid chat ID, bot removed from group, etc.):

Drop: Send Term disposition (skip forever, do not retry)
Alert: Publish a loop-guarded best-effort meta-alert to telegram.outbound itself (with subject telegram.alerts) notifying operators that a message dropped
Never crash: Continue processing next message

Loop guard on meta-alert: If the alert itself fails Terminal, do NOT generate a meta-alert-to-the-alert (infinite recursion guard).

Dedup Window

NATS stream TELEGRAM_OUTBOUND has a 5-minute duplicate_window keyed on Nats-Msg-Id (= msg_id). The connector does NOT deduplicate itself; NATS handles it.

Scenario: publisher publishes, network drops, publisher retries with same msg_id. NATS within the window returns the same message from stream (idempotent). Connector sees the same message twice, sends to Telegram twice, but:

Telegram has already processed the first send (message already in chat)
Second send is a duplicate in Telegram’s view (same logical message)
Telegram’s own idempotency may suppress it, or it shows twice in the chat (acceptable)

Note: Dedup at the Telegram API level (idempotent chat_id, text, message_id) is NOT guaranteed. This is acceptable — the connector’s job is to relay, not to deduplicate.

Rate Limiting Strategy

Telegram Bot API enforces a rate limit of ~30 messages per second per bot. Exceeding this returns 429 Too Many Requests with a Retry-After header (seconds).

Connector’s approach: Reactive, honor the header.

On 429 → extract Retry-After header (seconds)
Cap it to ack_wait (default 60s) to prevent indefinite blocking
Sleep min(retry_after, ack_wait)
Nak the message (negative-ack), let NATS re-deliver after nak delay
On re-delivery, try again

Why reactive, not proactive?

Proactive rate limiting (token bucket, sliding window) is complex and would require tracking every message’s send time
Reactive is simpler: let Telegram tell us when to back off
Dedup window ensures we don’t duplicate if we back off

Resilience: Consumer Reconnect

The outbound loop runs a resilient consumer that handles NATS stream-end and disconnects:

async fn run_resilient() {
  loop {
    match bind_pull_consumer(/* config */).await {
      Ok(consumer) => match run(consumer).await {
        Ok(()) => { /* stream-end, re-bind */ }
        Err(e) => { log::warn!("stream error, reconnect"); /* exponential backoff */ }
      }
      Err(e) => { log::warn!("bind failed, reconnect"); /* backoff */ }
    }
  }
}

On stream-end (all messages consumed, no new messages) → run() returns Ok(()) → loop re-binds the consumer
On NATS disconnect → error → backoff + re-bind
On bind failure → backoff + retry

This ensures the relay never exits due to transient NATS issues.

Implementation Details

MessageSource Trait Seam

trait MessageSource {
  async fn next(&mut self) -> Result<Option<IncomingMessage>>;
  async fn ack(&mut self, msg: &IncomingMessage) -> Result<()>;
  async fn nak(&mut self, msg: &IncomingMessage, delay: Option<Duration>) -> Result<()>;
  async fn term(&mut self, msg: &IncomingMessage) -> Result<()>;
}
 
pub struct IncomingMessage {
  pub data: Vec<u8>,
  pub headers: Headers,
  pub metadata: Metadata,
}

Impl: NatsSource (NATS JetStream pull consumer).

Main Relay Loop

async fn run(mut source: impl MessageSource) -> Result<()> {
  loop {
    match source.next().await? {
      None => return Ok(()), // stream-end
      Some(msg) => {
        match process_message(&msg).await {
          Ok(Disposition::Sent) => {
            source.ack(&msg).await?;
          }
          Ok(Disposition::Transient { retry_after }) => {
            source.nak(&msg, Some(retry_after)).await?;
          }
          Ok(Disposition::Terminal { alert_msg }) => {
            source.term(&msg).await?;
            if let Some(alert) = alert_msg {
              // best-effort publish alert
              nats_publish_alert(alert).await.ok();
            }
          }
          Err(e) => {
            log::error!("process error: {}", e);
            source.nak(&msg, None).await?; // retry
          }
        }
      }
    }
  }
}

TelegramClient: Classify Response

pub struct TelegramClient { /* ... */ }
 
impl TelegramClient {
  pub async fn send(&self, outbound: &TelegramOutbound) -> Result<Disposition> {
    let response = reqwest_send(/* ... */).await?;
    let status = response.status();
    
    match status {
      StatusCode::OK => {
        let body: SendMessageResponse = response.json().await?;
        Ok(Disposition::Sent)
      }
      StatusCode::TOO_MANY_REQUESTS => {
        let retry_after = parse_retry_after(&response)?;
        Ok(Disposition::Transient { retry_after })
      }
      StatusCode::BAD_REQUEST | StatusCode::FORBIDDEN => {
        let msg = response.text().await.unwrap_or_default();
        let alert = format_terminal_alert(&outbound, &msg);
        Ok(Disposition::Terminal { alert_msg: Some(alert) })
      }
      _ => Err(anyhow::anyhow!("unexpected status: {}", status))
    }
  }
}

Security: reqwest errors are stripped via without_url() — bot token lives in the request URL.

Graceful Shutdown

On SIGTERM signal:

Stop accepting new messages from the consumer
Wait for in-flight sends to complete (with timeout)
Close NATS connection
Exit 0

#[cfg(unix)]
async fn main() {
  let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
  
  tokio::spawn(async move {
    signal::ctrl_c().await.ok();
    shutdown_tx.send(()).ok();
  });
  
  run_outbound_relay(shutdown_rx).await?;
}

Configuration

Var	Default	Note
`TELEGRAM_BOT_TOKEN`	required	sole holder
`TELEGRAM_CHAT_ID`	required	default chat for `chat: None`
`NATS_URL`	required	fleet NATS spine
`NATS_TELEGRAM_NKEY_SEED`	required	NATS auth
`TG_DURABLE_NAME`	`tg_outbound`	consumer durable name
`TG_ACK_WAIT_SECS`	60	timeout before message re-delivery
`TG_MAX_DELIVER`	5	max times NATS will re-deliver before dropping
`TG_NAK_BACKOFF_SECS`	30	initial backoff on Nak
`META_ALERT_ON_DROP`	true	publish alert on Terminal
`TELEGRAM_API_BASE_URL`	official	override for testing

Testing

Unit Tests (trait-seamed)

#[test]
fn test_ack_on_sent() {
  // FakeTelegram returns Disposition::Sent
  // Verify ack() called on source
}
 
#[test]
fn test_nak_on_transient() {
  // FakeTelegram returns Disposition::Transient
  // Verify nak() called with retry_after
}
 
#[test]
fn test_term_on_terminal() {
  // FakeTelegram returns Disposition::Terminal
  // Verify term() called + alert published
}

Integration Tests (wiremock)

Mock the Telegram API and verify end-to-end behavior:

#[tokio::test]
async fn test_send_message_200() {
  let mock = wiremock_sendMessage(200);
  let result = client.send(&outbound).await;
  assert_eq!(result, Disposition::Sent);
}
 
#[tokio::test]
async fn test_rate_limit_429() {
  let mock = wiremock_sendMessage(429, "Retry-After: 45");
  let result = client.send(&outbound).await;
  assert!(matches!(result, Disposition::Transient { .. }));
}
 
#[tokio::test]
async fn test_bad_request_400() {
  let mock = wiremock_sendMessage(400, "Invalid chat_id");
  let result = client.send(&outbound).await;
  assert!(matches!(result, Disposition::Terminal { .. }));
}

Observability

Log discipline:

ERROR: 400/403 terminal sends (operator should investigate)
WARN: 429/5xx retries (informational)
INFO: consumer bind/reconnect transitions (state milestones)
DEBUG: per-message processing (behind RUST_LOG)

Structured attributes: nats_msg_id, chat, kind, status_code, retry_after, error_msg.

Telemetry: Spans for send_message, classify_response with timings + error status.

Future Work

Telegram Bot API v7 webhooks: Replace long-poll with push notifications (lower latency, fewer API calls)
Message dedup at API level: Extend TelegramOutbound with idempotency guarantees
Batch sends: Consume N messages, send as thread/replies instead of one-by-one
Per-chat rate limits: Track rate limits per chat, not globally

telegram-connector-complete — service overview
nats-jetstream-message-source — pull consumer implementation
task7-lib-bin-split-relay-loop-tests — relay loop + tests
security-reliability-fixes-bab38c4 — token leak fix

Levandor

Explorer

telegram_connector Outbound Relay — Design & Implementation

Outbound Relay — Durable JetStream → Telegram sendMessage

Design Principles

Message Flow

Wire Contract

Delivery Guarantees

At-Least-Once via Ack/Nak/Term

`Sent` → Ack (advance consumer offset)

`Transient` → Nak with backoff (retry)

`Terminal` → Term + drop + alert

Dedup Window

Rate Limiting Strategy

Resilience: Consumer Reconnect

Implementation Details

MessageSource Trait Seam

Main Relay Loop

TelegramClient: Classify Response

Graceful Shutdown

Configuration

Testing

Unit Tests (trait-seamed)

Integration Tests (wiremock)

Observability

Future Work

Graph View

Table of Contents

Levandor

Explorer

telegram_connector Outbound Relay — Design & Implementation

Outbound Relay — Durable JetStream → Telegram sendMessage

Design Principles

Message Flow

Wire Contract

Delivery Guarantees

At-Least-Once via Ack/Nak/Term

Sent → Ack (advance consumer offset)

Transient → Nak with backoff (retry)

Terminal → Term + drop + alert

Dedup Window

Rate Limiting Strategy

Resilience: Consumer Reconnect

Implementation Details

MessageSource Trait Seam

Main Relay Loop

TelegramClient: Classify Response

Graceful Shutdown

Configuration

Testing

Unit Tests (trait-seamed)

Integration Tests (wiremock)

Observability

Future Work

Related

Graph View

Table of Contents

`Sent` → Ack (advance consumer offset)

`Transient` → Nak with backoff (retry)

`Terminal` → Term + drop + alert