Help us shape the future of Go! Take the 2026 Go Developer Survey

transport

package
v1.10.0-alpha.21 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 28, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package transport defines the dependency-free observer contract used by Loom's generated HTTP, JSON-RPC, and Loom-MCP transports.

Enablement

Enablement is context-based. Applications attach an Observer to the request context via HTTPMiddleware (or WithObserver for non-HTTP entry points), and generated code resolves it transparently. Generated constructor signatures remain source-compatible with prior Loom releases — turning observability on or off is purely a wiring choice at the application boundary.

A minimal observer that prints reasons to stderr:

mw := transport.HTTPMiddleware(transport.ObserverFunc(func(_ context.Context, e transport.Event) {
    log.Printf("%s/%s %s reason=%s status=%d", e.Transport, e.Kind, e.Method, e.Reason, e.StatusCode)
}))
http.ListenAndServe(":8080", mw(generatedServer))

For non-HTTP entry points (e.g. a JSON-RPC consumer that reads frames from a Kafka topic), use WithObserver directly to inject the observer into the request-scoped context before invoking the generated handler.

Redaction rules

Generated code never emits raw bodies, JSON-RPC params, MCP tool arguments, credentials, or result payloads. Event only carries classified, low-cardinality fields safe for metric labeling and structured-log enrichment:

  • Transport (http/jsonrpc/mcp), Kind (start/finish/failure/stream*), and Reason are stable enumerations and may be used as metric labels.
  • Service, Method, Route, HTTPMethod identify the operation.
  • StatusCode, BytesWritten, Duration measure the response.
  • JSONRPCMethod, JSONRPCID, BatchCount, Notification are populated only after the JSON-RPC envelope has been decoded; pre-decode rejection events leave them empty by design.
  • SafeMessage carries operator-redacted text, never raw user input.

Observer implementations that need to extract more detail (request id, trace id, principal) should read from the request context inside Observer.ObserveEvent instead of expecting it on Event.

Stream lifecycle

SSE opening follows header commitment, even if the opening flush fails. Natural completion or terminal I/O failure closes a committed stream once; unopened and repeated closure emit no physical lifecycle event. Recoverable prewrite failures do not imply closure. Regenerate servers to adopt the shared stream binding and handler cleanup.

A WebSocket connection has one physical open/close pair. JSON-RPC operation bindings retain their decoded method and ID independently of later envelopes or per-call contexts. StreamClose with ReasonStreamFinalResponseSuppressed is a logical operation marker, not physical socket closure.

Callbacks execute outside stream ownership locks and may be concurrent or reentrant. Only one pending opening and closing record can be retained; repeated failure events remain synchronous. Background WebSocket cleanup records closure without invoking callbacks on a new background stack. Foreground operations or handler cleanup deliver it; a handler that ignores cancellation can delay that notification after native release. Observer panics propagate on the invoking stack, with an original handler panic taking precedence during cleanup.

Boundary with observability/otel

This package handles request and stream classification. OpenTelemetry span attributes, propagators, metric mode, and provider bootstrap remain in github.com/CaliLuke/loom/observability/otel. The two are composable: `otel.HTTPMiddleware` traces the request, `transport.HTTPMiddleware` classifies it. Stack them in any order; neither package depends on the other.

MCP release dependency

`loom-mcp` consumes this package through its `go.mod` `replace github.com/CaliLuke/loom => ../loom` directive during local development. A non-local `loom-mcp` release that drops the replace must bump `github.com/CaliLuke/loom` in `loom-mcp/go.mod` to a Loom tag that contains this package — otherwise generated MCP server code will not compile against the public Loom module.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func HTTPMiddleware

func HTTPMiddleware(observer Observer) func(http.Handler) http.Handler

HTTPMiddleware returns an HTTP middleware that injects observer into the request context so generated handlers can resolve it via ObserverFromContext. A nil observer disables event delivery without affecting the wrapped handler chain.

HTTPMiddleware is intentionally narrow: it only performs context injection. Span/trace setup, propagation, and metric recording remain the responsibility of `loom/observability/otel`, which can be composed on the outside of this middleware.

func Observe

func Observe(ctx context.Context, e Event)

Observe delivers e to the Observer carried by ctx, if any. Observe is the single entry point transport runtimes and generated adapters use to emit events so that a missing observer remains a cheap no-op.

func WithObserver

func WithObserver(ctx context.Context, o Observer) context.Context

WithObserver returns a copy of ctx that carries o. A nil observer clears any observer already attached to ctx.

func WithRequestObserver

func WithRequestObserver(ctx context.Context, obs *RequestObserver) context.Context

WithRequestObserver returns a copy of ctx that carries obs so deeper generated layers (JSON-RPC method dispatch, MCP tool handlers) can resolve the lifecycle observer without threading a new parameter through every signature.

func WithStreamObserver

func WithStreamObserver(ctx context.Context, observer *StreamObserver) context.Context

WithStreamObserver attaches immutable stream identity without changing ctx's cancellation lifetime. Retained operation wrappers use their own binding even when later writes receive a different per-call context.

Types

type BodyReadCapture

type BodyReadCapture struct {
	// contains filtered or unexported fields
}

BodyReadCapture observes bytes consumed downstream of its wrapped body. Body snapshots are safe during Read and Close. Underlying body Read/Close concurrency requirements remain unchanged.

func CaptureBodyReads

func CaptureBodyReads(body io.ReadCloser, limit int) (io.ReadCloser, *BodyReadCapture)

CaptureBodyReads returns a passive body wrapper and its separate capture state. Each Read and Close reaches the original body with its returned count and error unchanged. No reads, drains or closes occur on construction or observation. Nil and http.NoBody are returned unchanged as complete bodies. At most limit bytes are retained; exceeding it discards the entire preview without limiting downstream bytes. limit must be positive or this panics.

func (*BodyReadCapture) Body

func (c *BodyReadCapture) Body() ([]byte, bool, bool)

Body returns detached consumed bytes, whether the preview exceeded its limit, and whether EOF was observed (or the original body was nil/http.NoBody). Complete describes the read observation, not a Content-Length comparison; an error or Close does not imply completion. Snapshots never read the body.

type Event

type Event struct {
	// Kind is the lifecycle position of the event.
	Kind EventKind
	// Transport identifies the generated transport (HTTP, gRPC, JSON-RPC, MCP).
	Transport TransportKind
	// Reason is the stable classification for this event.
	Reason Reason

	// Service is the Loom service name owning the endpoint.
	Service string
	// Method is the operation name (HTTP handler, gRPC method, JSON-RPC
	// method, or MCP tool) as known to the transport.
	Method string
	// Route is the HTTP route pattern when available.
	Route string
	// HTTPMethod is the HTTP verb when the underlying transport is HTTP.
	HTTPMethod string
	// StatusCode is the final HTTP status code when known.
	StatusCode int
	// Duration is the wall-clock duration between request start and the
	// emission of the terminal event.
	Duration time.Duration
	// BytesWritten counts bytes flushed to the response writer when known.
	BytesWritten int64

	// JSONRPCMethod is the JSON-RPC method name decoded from the request
	// envelope.
	JSONRPCMethod string
	// JSONRPCID is the decoded JSON-RPC request id rendered as a string.
	JSONRPCID string
	// BatchCount is the number of envelopes in a JSON-RPC batch request,
	// zero for non-batch requests.
	BatchCount int
	// Notification reports whether the JSON-RPC request had no id and is a
	// notification.
	Notification bool

	// SessionID is the transport session identifier when known. For MCP
	// this is the Mcp-Session-Id; for WebSocket and SSE streams it is the
	// stream session id.
	SessionID string
	// RequestID is the generator-assigned request correlation id when
	// known.
	RequestID string
	// TraceID is the propagated trace id when known.
	TraceID string

	// ErrorClass categorizes the error using a short, redaction-safe
	// label.
	ErrorClass string
	// SafeMessage is a redaction-safe error message. Generated code never
	// places raw error messages from untrusted sources in this field.
	SafeMessage string

	// Attrs carries additional redaction-safe key/value attributes. Keys
	// and values are intended to be safe for log enrichment and metric
	// labeling; raw payloads must not be placed here.
	Attrs map[string]string
}

Event is the payload delivered to an Observer. All fields are optional; generators populate the fields that are well defined at the point where the event is emitted and leave the rest zero. Generated code never populates Event with raw bodies, JSON-RPC params, MCP tool arguments, credentials, or result payloads.

type EventKind

type EventKind uint8

EventKind identifies the lifecycle position a transport event reports.

const (
	// EventKindRequestStart marks the beginning of a request lifecycle.
	EventKindRequestStart EventKind = iota + 1
	// EventKindRequestFinish marks a successful terminal request event.
	EventKindRequestFinish
	// EventKindRequestFailure marks a failed terminal request event.
	EventKindRequestFailure
	// EventKindStreamOpen marks the opening of a streaming response.
	EventKindStreamOpen
	// EventKindStreamClose marks the closing of a streaming response.
	EventKindStreamClose
	// EventKindStreamFailure marks a failed stream write or flush.
	EventKindStreamFailure
)

type Observer

type Observer interface {
	// ObserveEvent is invoked with a context that is alive for the
	// request producing the event. Implementations must not retain ctx
	// past the call, and must treat e as read-only.
	ObserveEvent(ctx context.Context, e Event)
}

Observer receives framework transport lifecycle events. Observer implementations must be safe for concurrent use.

func ObserverFromContext

func ObserverFromContext(ctx context.Context) Observer

ObserverFromContext returns the Observer attached to ctx, or nil if none has been injected.

type ObserverFunc

type ObserverFunc func(ctx context.Context, e Event)

ObserverFunc adapts an ordinary function into an Observer.

func (ObserverFunc) ObserveEvent

func (f ObserverFunc) ObserveEvent(ctx context.Context, e Event)

ObserveEvent calls f with the supplied event. A nil ObserverFunc is a no-op.

type Reason

type Reason string

Reason is a stable, low-cardinality classification of why an event was emitted. Reason values are part of the package's public contract and are safe to use as metric labels.

const (
	// ReasonOK marks a terminal event with no error classification.
	ReasonOK Reason = "ok"
	// ReasonRequestDecodeFailed marks failure to decode the HTTP request body
	// or query parameters into the generated request type.
	ReasonRequestDecodeFailed Reason = "request_decode_failed"
	// ReasonInvalidJSONRPCEnvelope marks failure to parse the JSON-RPC
	// envelope.
	ReasonInvalidJSONRPCEnvelope Reason = "invalid_jsonrpc_envelope"
	// ReasonInvalidJSONRPCBatch marks rejection of a malformed JSON-RPC batch
	// request.
	ReasonInvalidJSONRPCBatch Reason = "invalid_jsonrpc_batch"
	// ReasonInvalidJSONRPCMethod marks rejection of a JSON-RPC envelope whose
	// method field is missing or malformed.
	ReasonInvalidJSONRPCMethod Reason = "invalid_jsonrpc_method"
	// ReasonInvalidJSONRPCParams marks failure to decode JSON-RPC params into
	// the generated parameter type.
	ReasonInvalidJSONRPCParams Reason = "invalid_jsonrpc_params"
	// ReasonUnsupportedMethod marks rejection of a JSON-RPC method the
	// generated handler does not implement.
	ReasonUnsupportedMethod Reason = "unsupported_method"
	// ReasonMissingCredentials marks rejection because required credentials
	// were not presented.
	ReasonMissingCredentials Reason = "missing_credentials"
	// ReasonInvalidCredentials marks rejection because credentials were
	// presented but could not be validated.
	ReasonInvalidCredentials Reason = "invalid_credentials"
	// ReasonPermissionRejected marks rejection because authorization checks
	// denied the principal.
	ReasonPermissionRejected Reason = "permission_rejected"
	// ReasonPrincipalMismatch marks rejection because the principal did not
	// match the session owner.
	ReasonPrincipalMismatch Reason = "principal_mismatch"
	// ReasonHandlerError marks a handler returning a non-nil error.
	ReasonHandlerError Reason = "handler_error"
	// ReasonPanic marks a recovered panic in the generated handler.
	ReasonPanic Reason = "panic"
	// ReasonResponseWriteFailed marks a failure while encoding a response or
	// writing response metadata or body data.
	ReasonResponseWriteFailed Reason = "response_write_failed"
	// ReasonStreamWriteFailed marks a failed write to a streaming response.
	ReasonStreamWriteFailed Reason = "stream_write_failed"
	// ReasonStreamFlushFailed marks a failed flush of a streaming response.
	ReasonStreamFlushFailed Reason = "stream_flush_failed"
	// ReasonStreamWriteTimeout marks a timed-out streaming response write.
	ReasonStreamWriteTimeout Reason = "stream_write_timeout"
	// ReasonStreamFlushTimeout marks a timed-out streaming response flush.
	ReasonStreamFlushTimeout Reason = "stream_flush_timeout"
	// ReasonStreamFinalResponseSuppressed marks a SendAndClose final value
	// discarded because the stream carries no JSON-RPC request ID (a
	// notification or a raw GET events/stream listener), so protocol rules
	// forbid sending a final response.
	ReasonStreamFinalResponseSuppressed Reason = "stream_final_response_suppressed"
	// ReasonMCPSessionMissing marks an MCP events-stream request that omitted
	// the Mcp-Session-Id header.
	ReasonMCPSessionMissing Reason = "mcp_session_missing"
	// ReasonMCPSessionNotFound marks an MCP events-stream request whose
	// session id is unknown to the server.
	ReasonMCPSessionNotFound Reason = "mcp_session_not_found"
	// ReasonMCPSessionPrincipalMismatch marks an MCP events-stream request
	// whose authenticated principal differs from the session owner.
	ReasonMCPSessionPrincipalMismatch Reason = "mcp_session_principal_mismatch"
	// ReasonMCPEventsStreamWriteFailed marks a failure writing an MCP
	// notification frame to the events-stream response.
	ReasonMCPEventsStreamWriteFailed Reason = "mcp_events_stream_write_failed"
)

type RequestObserver

type RequestObserver struct {
	// contains filtered or unexported fields
}

RequestObserver tracks a single transport request lifecycle and emits Start, terminal, and panic events through an Observer attached to the request context. HTTP, gRPC, JSON-RPC, and MCP transports use the observer to consolidate per-request boilerplate into a small surface:

obs, w := transport.BeginHTTPRequest(ctx, w, "Service", "Method", r)
defer obs.End()
// ...
obs.Fail(transport.ReasonRequestDecodeFailed) // before an error return

End must be deferred so panics propagate through it as EventKindRequestFailure with ReasonPanic and are re-panicked; if End is reached without a recorded failure, EventKindRequestFinish is emitted. RequestObserver methods are safe for concurrent use.

func BeginHTTPRequest

func BeginHTTPRequest(ctx context.Context, w http.ResponseWriter, service, method string, r *http.Request) (*RequestObserver, http.ResponseWriter)

BeginHTTPRequest wraps w with a captured response writer, records the request start time, emits an EventKindRequestStart event, and returns the observer together with the wrapped writer. Generated HTTP handlers reassign their response writer parameter to the returned writer so all subsequent writes are captured.

func BeginJSONRPCRequest

func BeginJSONRPCRequest(ctx context.Context, w http.ResponseWriter, service string, r *http.Request) (*RequestObserver, http.ResponseWriter)

BeginJSONRPCRequest wraps w with a captured response writer and starts a request lifecycle classified as TransportJSONRPC. service identifies the Loom service; the JSON-RPC envelope's method, id, batch count, and notification flag are filled in later through RequestObserver.SetJSONRPC once the envelope has been decoded. Pre- decode rejection events therefore leave the JSON-RPC fields empty as the plan requires.

func BeginRequest

func BeginRequest(ctx context.Context, kind TransportKind, service, method string) *RequestObserver

BeginRequest starts a transport request lifecycle without an HTTP response writer. JSON-RPC and MCP generators that need to emit start events outside the HTTP capture pipeline use this entry point. End and Fail behave identically to the HTTP variant; StatusCode and BytesWritten remain zero unless updated through ApplyHTTPStatus.

func RequestObserverFromContext

func RequestObserverFromContext(ctx context.Context) *RequestObserver

RequestObserverFromContext returns the RequestObserver injected by WithRequestObserver, or nil if none is attached. Generated code uses nil-safe methods on the returned observer so a missing observer remains a cheap no-op.

func (*RequestObserver) EmitStreamClose

func (o *RequestObserver) EmitStreamClose()

EmitStreamClose reports the natural close of a streaming response.

func (*RequestObserver) EmitStreamFailure

func (o *RequestObserver) EmitStreamFailure(reason Reason)

EmitStreamFailure reports a failed write or flush during streaming. The terminal request event is unaffected; callers that want the failure to also classify the request must call Fail with the same reason.

func (*RequestObserver) EmitStreamOpen

func (o *RequestObserver) EmitStreamOpen()

EmitStreamOpen reports the opening of a streaming response within the current request. Callers own deduplication and request-span lifetime. SSE and WebSocket runtimes use StreamObserver for retained physical lifecycle state.

func (*RequestObserver) End

func (o *RequestObserver) End()

End emits the terminal request event. End must be deferred so a recovered panic emits EventKindRequestFailure with ReasonPanic before re-panicking. End is a no-op on a nil observer or after the first call, allowing it to be safely composed with explicit failure emission.

func (*RequestObserver) Fail

func (o *RequestObserver) Fail(reason Reason)

Fail records reason as the terminal classification for the request. The first call wins so the originating failure is not overwritten by downstream cleanup classifications. End will emit EventKindRequestFailure with the recorded reason.

func (*RequestObserver) FailWithMessage

func (o *RequestObserver) FailWithMessage(reason Reason, safeMessage string)

FailWithMessage records reason and a redaction-safe message for the terminal event. The message is reported through Event.SafeMessage and must not contain raw user input, credentials, or payload data.

func (*RequestObserver) ResponseCommitted

func (o *RequestObserver) ResponseCommitted() bool

ResponseCommitted reports whether the captured HTTP writer has committed a final status. It is false for an observer without an HTTP writer.

func (*RequestObserver) SetJSONRPC

func (o *RequestObserver) SetJSONRPC(method, id string, batchCount int, notification bool)

SetJSONRPC populates JSON-RPC-specific fields on the terminal event. It is safe to call once the JSON-RPC envelope has been decoded. When the observer's operation name was not provided at Begin time, method is also used as the observer's Method so the terminal event reports the JSON-RPC method as the operation name.

func (*RequestObserver) SetSession

func (o *RequestObserver) SetSession(sessionID string)

SetSession records the transport session identifier carried by the request, when known.

type ResponseCapture

type ResponseCapture struct {
	// contains filtered or unexported fields
}

ResponseCapture records response status, byte counts and an optional bounded body preview. Its accessors are safe to call concurrently with response writes. The wrapped ResponseWriter retains its own concurrency requirements.

func CaptureResponse

CaptureResponse returns a writer and its separate capture state. The writer preserves the optional interfaces supported by httpsnoop and exposes Unwrap for http.ResponseController. Informational responses do not commit the final status, except 101 (Switching Protocols). Writes and flushes commit an implicit 200; a ReadFrom that transfers no bytes does not commit a response. By default no body bytes or headers are retained; options enable bounded body capture and first-observed-commit header snapshots. Opaque ReadFrom commits cannot establish header timing unless an earlier operation already committed.

func (*ResponseCapture) Body

func (c *ResponseCapture) Body() ([]byte, bool)

Body returns a detached copy of accepted response bytes and whether capture was omitted after exceeding its configured limit. Without body capture it returns nil, false. It is safe to call concurrently with response writes.

func (*ResponseCapture) BytesWritten

func (c *ResponseCapture) BytesWritten() int64

BytesWritten returns the actual bytes transferred through Write, WriteString and ReadFrom, including partial writes that return an error.

func (*ResponseCapture) Headers

func (c *ResponseCapture) Headers() (http.Header, bool)

Headers returns a detached first-commit snapshot and whether it is available. It returns nil, false without the option, before an observed commit, or when forwarding changed headers or ReadFrom committed first. It is safe to call concurrently with response writes; it never reads the writer's live Header map.

func (*ResponseCapture) StatusCode

func (c *ResponseCapture) StatusCode() int

StatusCode returns the first final response status or zero before any response is committed. Hijacked connection writes are outside the capture.

type ResponseCaptureOption

type ResponseCaptureOption func(*ResponseCapture)

ResponseCaptureOption configures optional response observation before any writes occur. CaptureResponse retains no body unless an option enables it.

func WithResponseBodyCapture

func WithResponseBodyCapture(limit int) ResponseCaptureOption

WithResponseBodyCapture retains at most limit accepted body bytes. If the response exceeds limit, the entire preview is discarded and Body reports omission. The complete response still reaches the underlying writer. limit must be positive; otherwise this function panics. Hijacked I/O is not captured.

func WithResponseHeaderCapture

func WithResponseHeaderCapture() ResponseCaptureOption

WithResponseHeaderCapture snapshots headers surrounding the first final WriteHeader, Write, WriteString or Flush call. The snapshot is available only when forwarding leaves the headers unchanged. Later handler mutations do not alter the snapshot. An opaque ReadFrom that commits first makes headers unavailable: its internal read/write timing cannot establish which headers it sent. The underlying writer retains its normal concurrency requirements.

type StreamObserver

type StreamObserver struct {
	// contains filtered or unexported fields
}

StreamObserver retains immutable request identity independently of the request span. Transport owners record physical lifecycle transitions under their own locks, then call Drain outside those locks. At most one opening and one closing event are retained; failures are delivered synchronously.

func NewStreamObserver

func NewStreamObserver(ctx context.Context, kind TransportKind) *StreamObserver

NewStreamObserver snapshots the request identity attached to ctx. If none is attached, kind identifies the transport and other identity fields stay empty. Ending or changing the request observer does not change this binding.

func StreamObserverFromContext

func StreamObserverFromContext(ctx context.Context, kind TransportKind) *StreamObserver

StreamObserverFromContext returns a retained binding, or snapshots the request observer when no binding is attached. It never changes a connection binding.

func (*StreamObserver) Drain

func (s *StreamObserver) Drain()

Drain synchronously delivers pending lifecycle events outside all observer locks. Concurrent or reentrant drains return; the active drainer delivers any newly recorded closure after opening. A callback panic propagates and releases the drain claim, leaving remaining metadata available to foreground cleanup.

func (*StreamObserver) Failure

func (s *StreamObserver) Failure(reason Reason)

Failure synchronously reports a write/flush failure without changing physical lifecycle or request classification. Call it outside transport ownership locks.

func (*StreamObserver) ForContext

func (s *StreamObserver) ForContext(ctx context.Context) *StreamObserver

ForContext selects an explicitly retained operation binding from ctx, falling back to this connection binding. It does not infer identity from arbitrary inbound request observers carried into an outgoing client operation.

func (*StreamObserver) JSONRPC

func (s *StreamObserver) JSONRPC(method, id string, notification bool) *StreamObserver

JSONRPC returns a separate immutable operation binding. It shares connection identity and the observer, but not physical lifecycle state or method identity.

func (*StreamObserver) RecordClose

func (s *StreamObserver) RecordClose()

RecordClose records physical closure only after a recorded opening. Closing an unopened stream and repeated closure do not produce events.

func (*StreamObserver) RecordOpen

func (s *StreamObserver) RecordOpen()

RecordOpen records successful stream commitment once, without calling user code. A later Drain delivers the event on its caller's stack.

func (*StreamObserver) Suppressed

func (s *StreamObserver) Suppressed()

Suppressed reports the existing logical final-response suppression marker. It does not claim or deduplicate the physical connection's closing event.

type TransportKind

type TransportKind string

TransportKind identifies the generated transport that produced an event.

const (
	// TransportHTTP identifies generic HTTP-generated transport events.
	TransportHTTP TransportKind = "http"
	// TransportGRPC identifies gRPC-generated transport events.
	TransportGRPC TransportKind = "grpc"
	// TransportJSONRPC identifies JSON-RPC-generated transport events.
	TransportJSONRPC TransportKind = "jsonrpc"
	// TransportMCP identifies Loom-MCP-generated transport events.
	TransportMCP TransportKind = "mcp"
)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL