Streaming ingestion

Streaming is the write path the Python SDK uses. A process registers itself as a logger on a run, then keeps one bidirectional gRPC stream open for the life of the run. Each series is declared once per stream and then addressed by a small numeric id, so updates carry only points. Streaming is gRPC only.

client                                   server
  │ InitializeRunLogging(run, logger_id)    │
  │────────────────────────────────────────▶│  create or find the run, register the logger
  │◀──────────── InitializedRun ────────────│  run_id, created?
  │◀──────────── LastLoggedSteps … ─────────│  what the run already holds
  │◀──────────── RlFamilyDeclarations … ────│  (RL runs only)
  │◀──────────── EnvStatuses (is_last) ─────│  handshake complete; stream ends
  │                                         │
  │ UpdateRunStream                         │
  │═══ UpdateRunRequest #0 (declarations) ═▶│
  │═══ UpdateRunRequest #1 ════════════════▶│
  │◀══════════════ ack #0 ══════════════════│
  │◀══════════════ ack #1 ══════════════════│
  │   …                                     │
  │═══ UpdateRunRequest #n (close_run) ════▶│
  │◀══════════════ ack #n ══════════════════│  run closed once every logger has closed

Choosing a logger id

logger_id is a positive 64-bit number identifying one writer process on one run. Choose it randomly when the process starts and use it for the handshake and every stream message for that run. Several processes writing to one run (for example, the ranks of a distributed job) each use their own. Metrana uses the id to recognize retried data, so a restarted process must choose a new one rather than reuse its predecessor’s.

1. The handshake: InitializeRunLogging

InitializeRunLogging takes the same run definition as CreateRun, plus the logger id and a resume policy:

{
  "run": {"workspace": "my-workspace", "project": "vision", "runName": "ppo-seed3", "orchestrationIdentifier": "slurm-8812345"},
  "createProjectIfMissing": true,
  "loggerId": "5813377020184215",
  "resumePolicy": "RESUME_POLICY_NEVER"
}

It creates the run if it does not exist, registers the logger, waits until the ingestion pipeline has caught up with anything already written to the run, and streams back what the run holds. Messages arrive in this order:

1. initializedRun — exactly one. runId (16 bytes) addresses the run on the update stream. created is true when this logger created the run just now.
2. lastLoggedSteps — one or more pages; the last has isLast: true. The last step of every non-RL series in the run, with its ordering and value kind. Labels are given as key=value pairs joined by &. Use these to continue step counters when resuming.
3. rlFamilyDeclarations — zero or more pages; sent only for runs with RL series. Each RL metric’s label keys, ordering, value kind and whether it is episodic, so a resuming client can keep new writes consistent.
4. envStatuses — one or more pages; the last has isLast: true. The last RL step and episode of every environment.

The handshake is complete when envStatuses with isLast: true arrives; the server then ends the stream. A stream that ends with an error status did not complete. Waiting for the pipeline can take a few seconds on a busy deployment; after about 30 seconds the call fails with UNAVAILABLE and can be retried.

Resuming an existing run

When the run already exists, resumePolicy decides whether this logger may attach to it:

Policy	Attaches when
RESUME_POLICY_NEVER (and the default, UNSPECIFIED)	This logger id created the run, or the run’s stored orchestrationIdentifier equals the one in the request.
RESUME_POLICY_ALLOW	Always.

Under NEVER, a refusal still sends initializedRun with created: false and the run’s stored orchestrationIdentifier and creationTime, then ends the call with ALREADY_EXISTS. Nothing is registered, so the existing run is left untouched. This is what stops a new job that reuses a run name from silently appending to an old run, while letting every process of one job — which share an orchestration identifier — attach to the run whichever of them created it.

2. The stream: UpdateRunStream

Open UpdateRunStream with the authorization metadata. Every UpdateRunRequest carries the run_id from the handshake and your logger_id; neither may change within a stream. The first message fails with INVALID_ARGUMENT if the logger has not completed InitializeRunLogging.

Declaring series

Instead of repeating a series’ name, scale and labels in every update, declare it once with a SeriesDeclaration and refer to it by declaration_id:

{
  "runId": "AAAAAAAAAAAAAAAAAAAAAQ==",
  "loggerId": "5813377020184215",
  "seriesDeclarations": [
    {"declarationId": "1", "metricName": "train/loss", "ordering": "ORDERING_MODE_ABSOLUTE_ORDERED"},
    {"declarationId": "2", "metricName": "reward", "scale": "ENVIRONMENT_STEP",
     "ordering": "ORDERING_MODE_ABSOLUTE_ORDERED"}
  ],
  "series": [
    {"seriesId": "1", "pointBatches": [{"firstStep": "500", "values": [0.412, 0.409]}]}
  ],
  "rlSeries": [
    {"seriesId": "2", "environmentId": "env-07",
     "rlStepBatches": [{"rlStep": "500", "episode": "12", "firstMinorStep": "0", "values": [0.0, 1.0]}]}
  ]
}

(JSON is shown for readability; the stream carries protobuf.)

• Declarations may ride in the same message as their first use, or any earlier one.
• Declaration ids are yours to choose and belong to the stream: after a reconnect, declare again.
• Declaring an id again with identical content is accepted, so a replayed message is harmless. Declaring it again with different content, or using an undeclared id, fails the stream with INVALID_ARGUMENT.
• The declaration’s ordering applies to float and RL series; histogram and scatter series ignore it.

The update fields mirror UpdateRuns with declarations in place of identities: series (StreamingSeriesUpdate), histogramSeries, scatterSeries (which carry is3d and metadataLabels themselves) and rlSeries (which carry environmentId), alongside attributes and envAttrUpdates. Validation and limits are the same.

Acknowledgments

Every request receives exactly one UpdateRunResponse, in the order the requests were sent. An acknowledgment means the request was validated and accepted. (Its operation_id is currently always empty.)

Pipeline requests: send the next one without waiting for the previous acknowledgment. The server works on several at once and still applies them in order. Keep a bounded window of unacknowledged requests, and keep a copy of each until it is acknowledged — you need them to recover from a broken stream.

When the stream breaks

If a request fails, the server ends the stream with an error status. The failed request and every request after it may have been applied partially, or not at all. Requests acknowledged before the failure are safe.

To recover:

1. For INVALID_ARGUMENT, PERMISSION_DENIED or NOT_FOUND, stop: retrying the same data fails the same way.
2. Otherwise, open a new UpdateRunStream with the same run_id and logger_id. There is no need to repeat InitializeRunLogging.
3. Declare your series again.
4. Resend every unacknowledged request, in order.

Delivery is at-least-once: replayed points are deduplicated or overwrite themselves as described in Retries and duplicates.

The same applies when the connection drops, and when the stream ends without an error before you closed it: the server ends open streams cleanly when it restarts, for example during an upgrade. Treat an early end as a broken stream.

Back off exponentially between reconnects. RESOURCE_EXHAUSTED means a message did not fit under your workspace’s or organization’s write rate: wait longer before retrying and send less per second. It is not an outage, so do not treat it as a reason to fail over elsewhere; the SDK retries it in place with smaller requests.

3. Closing

Put closeRun on the logger’s last message — it may carry data too:

{"runId": "…", "loggerId": "5813377020184215", "closeRun": {"closeMessage": "finished"}}

It is applied after every earlier request. Once it is acknowledged, half-close the stream; the server then ends it. When every logger of the run has closed, the run’s status becomes closed.

A closed logger cannot stream again; initialize a new logger to write more. Re-sending the close message itself is accepted, so a close whose acknowledgment was lost can be retried safely.

A logger whose stream stays down for a long time without closing is marked crashed (after 15 minutes by default), and the run follows once no logger is running. Reconnecting afterwards still works.

Transport notes

• Messages are limited to 4 MiB. Batching many series and batches into one message is efficient; split before the limit.
• The server accepts zstd-compressed requests (grpc-encoding: zstd), which shrink float data considerably.
• The access token is checked when the stream opens, and an open stream outlives the token. Refresh before reconnecting.