Documentation
¶
Overview ¶
Package capture implements the lossy-channel + drainer pattern from ARCHITECTURE § 7.2 and the per-event timing capture from § 7.1 step 2.
The forwarding path teas every chunk through a Sink. Sinks are non-blocking writers: a slow disk or a saturated channel never throttles the forwarding read. Bytes that cannot be sent on the channel are dropped and the sink's onDrop callback fires (which flips truncated_req / _resp).
Channel payloads are Chunk values that carry a wire-arrival timestamp taken at Write() call time. The drainer is a passive consumer: when it later finds an SSE frame boundary in the byte stream, the relevant t_delta_ms is taken from the chunk that contained the first byte past the boundary, so the timing reflects when bytes arrived from upstream, not when the drainer woke up.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func LookupChunkTime ¶
func LookupChunkTime(timings []ChunkTiming, offset int64) (time.Time, bool)
LookupChunkTime returns the wall-clock arrival time of the chunk that contained the byte at the given file offset, or the zero time if no such chunk exists.
Used at finalize to attach t_delta_ms to each SSE event. Callers should pre-sort timings by offset; the drainer already does this because chunks are processed in order. Binary search; O(log N).
func ReleaseChunk ¶
func ReleaseChunk(c Chunk)
ReleaseChunk returns a Chunk's backing buffer to the pool. The drainer calls this once it has finished consuming the chunk.
Types ¶
type Chunk ¶
type Chunk struct {
// At is the wall-clock time the proxy received this chunk from upstream
// (or from client, for request capture).
At time.Time
// Data is a private copy of the bytes; callers may reuse their buffer
// after Write() returns.
Data []byte
}
Chunk is one byte payload on a capture channel, tagged with the time captureSink.Write was called (= time the inner Transport delivered these bytes to the proxy = wire-arrival time, modulo Go transport buffering).
type ChunkTiming ¶
type ChunkTiming struct {
Offset int64 // file offset where this chunk's bytes start
At time.Time // wire-arrival time recorded by Sink.Write
}
ChunkTiming pairs a tmp-file byte offset with the wire-arrival time of the chunk that produced those bytes. The drainer records one per successfully written chunk so finalize can look up: "which chunk contained the first byte of this SSE event?" → its timestamp.
type DrainResult ¶
type DrainResult struct {
// BytesWritten is the total bytes written to the tmp file. Equals the
// uncompressed Content-Length for clean captures; less if Truncated.
BytesWritten int64
// Truncated is true if max_body_bytes was exceeded; further bytes were
// silently dropped after the cap. Note: this is distinct from the
// channel-level drop tracked by Sink.OnDrop — both flip truncated_*
// on the trace, but for different reasons.
Truncated bool
// FirstByteAt is the timestamp from the first non-empty chunk seen, or
// zero if no chunks arrived. Used for sanity / debugging; not in JSONL.
FirstByteAt int64 // unix ns
// Err is the first non-EOF write error, if any. The drain itself does
// not abort on disk errors — it stops writing but keeps draining the
// channel so the producer can complete.
Err error
// ChunkTimings records the (file offset, arrival time) of each
// non-empty chunk that was written. Used by the finalize phase to
// attach per-event t_delta_ms to SSE events per ARCHITECTURE § 3.3.
//
// Chronologically ordered (offsets monotonically increase). Empty
// for traces that had no bytes (or only dropped bytes).
ChunkTimings []ChunkTiming
}
DrainResult is the outcome of a single capture direction's drain loop. Populated by Drain() and consumed by the trace's finalize step.
func Drain ¶
func Drain(ch <-chan Chunk, tmpFile io.Writer, maxBodyBytes int64) DrainResult
Drain reads chunks from ch and appends their bytes to tmpFile, enforcing the maxBodyBytes cap. Returns when ch is closed.
The drain never blocks the producer (ch is bounded; producers send non-blockingly via Sink). If disk writes fail mid-stream, drain stops writing but keeps reading the channel to EOF — this guarantees the producer's non-blocking sends keep finding the channel "open enough" to drop or send.
type Sink ¶
type Sink struct {
// Ch is the capture channel. Closed by the owner (the forwarding
// goroutine) at finalize; Sink.Write must not be called after Close.
Ch chan<- Chunk
// OnDrop is called for every Write that dropped its bytes. Typically
// flips a per-direction truncated flag plus bumps a counter.
OnDrop func()
// OnByte is called once per non-empty Write — *before* the channel
// send — so a stream-idle watchdog can pulse without depending on
// the chunk reaching the drainer. Optional; nil = no callback.
OnByte func()
// Now is overridable for tests; production keeps it nil → time.Now.
Now func() time.Time
}
Sink is a non-blocking writer that copies each Write into a fresh slice and tries a non-blocking send on its channel. If the channel is full the bytes are dropped and OnDrop is invoked (once per dropped Write).
func (*Sink) Write ¶
Write copies p, stamps the current time, and non-blockingly sends the chunk on s.Ch. Always returns (len(p), nil) — Sinks cannot fail in the io.Writer sense; the forwarding goroutine's reads are never short-read or error'd because of capture behavior.
The byte slice is taken from a sync.Pool; the drainer must call ReleaseChunk(c) once it has finished consuming the chunk so the backing array returns to the pool.
type TmpDir ¶
type TmpDir struct {
Path string
}
TmpDir manages the data/tmp/ directory documented in ARCHITECTURE § 2. On startup the entire directory is wiped (orphans from a prior crash).
func NewTmpDir ¶
NewTmpDir creates and wipes the tmp directory under dataDir. Must be called exactly once at process startup, before any traces are captured.
func (*TmpDir) CreateTraceFiles ¶
CreateTraceFiles opens two new files (req + resp) for the given trace ID. The caller closes them at finalize.
func (*TmpDir) RemoveTraceFiles ¶
RemoveTraceFiles deletes a trace's tmp files. Called by the writer goroutine after the JSONL line and SQLite row are durably written. Errors are non-fatal: orphan tmp files will be cleaned at next startup.
Source Files