Skip to main content

Ingest telemetry

A production-shaped ingest loop for POST /telemetry. The contract it respects:

  • The body is a bare JSON array of records; at most 1,000 records per batch (413 batch_too_large beyond that) and at most 1 MB per body.
  • A 202 acknowledgment reports accepted_count, rejected_count, and rejected_indices. Accepted rows are already written, so on partial rejection the loop repairs only the rejected records and never resubmits the whole batch.
  • 429 and 503 responses were rejected or failed cleanly and are retried after Retry-After. A 401 stops immediately (retrying a bad key trips the auth lockout).
  • Timeouts and dropped connections are ambiguous: the batch may have landed, and ingest has no dedup, so these loops surface the error instead of blind-retrying. See the idempotency warning.

Set CONSTELLATION_API_BASE and CONSTELLATION_API_TOKEN first (common setup).

const BASE = process.env.CONSTELLATION_API_BASE ?? 'https://api.constellation.space'
const TOKEN = process.env.CONSTELLATION_API_TOKEN
if (!TOKEN) throw new Error('CONSTELLATION_API_TOKEN is not set')

const BATCH_MAX = 1000

type TelemetryRecord = {
  measurement: string
  time: string
  tags?: Record<string, string>
  fields?: Record<string, number | string>
}

type Ack = {
  write_id: string
  accepted_count: number
  rejected_count: number
  rejected_indices: number[] | null
}

async function postBatch(records: TelemetryRecord[], attempt = 0): Promise<Ack> {
  const res = await fetch(`${BASE}/telemetry`, {
    method: 'POST',
    headers: { 'x-api-key': TOKEN!, 'content-type': 'application/json' },
    body: JSON.stringify(records),
  })
  if (res.status === 429 || res.status === 503) {
    if (attempt >= 5) throw new Error(`giving up after ${attempt} retries (${res.status})`)
    const wait = Number(res.headers.get('retry-after') ?? '5')
    await new Promise((r) => setTimeout(r, wait * 1000))
    return postBatch(records, attempt + 1)
  }
  if (res.status === 401) throw new Error('API key rejected (401); stop before tripping lockout')
  if (res.status !== 202) throw new Error(`telemetry write failed: ${res.status} ${await res.text()}`)
  return (await res.json()) as Ack
}

async function ingest(records: TelemetryRecord[]) {
  for (let i = 0; i < records.length; i += BATCH_MAX) {
    const batch = records.slice(i, i + BATCH_MAX)
    const ack = await postBatch(batch)
    console.log(`write ${ack.write_id}: accepted ${ack.accepted_count}, rejected ${ack.rejected_count}`)
    for (const idx of ack.rejected_indices ?? []) {
      // Accepted rows are already written; repair and resubmit only these.
      console.error(`rejected record ${idx}:`, JSON.stringify(batch[idx]))
    }
  }
}

const now = new Date().toISOString()
await ingest([
  {
    measurement: 'satellite',
    time: now,
    tags: { node_id: 'SAT-0012', node_type: 'satellite' },
    fields: { lat: 47.61, lon: -122.33, altitude_km: 550.2, utilization: 0.42 },
  },
  {
    measurement: 'link',
    time: now,
    tags: { link_id: 'SAT-0012:GS-SEA-01', node_id: 'SAT-0012' },
    fields: { snr_db: 12.4, utilization: 0.55 },
  },
])

Expected output for a clean write:

write w-8c41f2ae: accepted 2, rejected 0