bridge
import "github.com/danmestas/dagnats/bridge"bridge/journal.go The agent journal over the HTTP bridge (ADR-026). A non-Go worker claims a task’s journal with its first journal.append or journal.read, and from then on its complete, fail, continue and pause commit through the same journal package a Go worker uses, so the fencing is the package’s and not a copy of it. The claim carries the delivery count of the polled message: a redelivery to a second poller claims with a higher count and fences this one.
Outcomes the worker must handle, mirrored from the Go SDK’s errors:
409 {"code":"fenced"} another claim follows this worker's. The
task is the successor's: stop, write and
resolve nothing more.
410 {"code":"stale"} a later iteration holds the step. The task
was acknowledged; stop.
410 {"code":"committed"} this attempt already committed. The bridge
republished its lifecycle event and
acknowledged the task; stop.
503 {"code":"not_provisioned"} the AGENT_JOURNAL stream is missing.bridge/logs.go POST /v1/tasks/{id}/logs (#624): the HTTP-bridge counterpart to the worker SDK’s LogOut()/LogErr() — lets a non-Go worker publish stdout/stderr-tagged chunks to the BUILD_LOGS hot lane. Ownership is enforced exactly like resolve (authorizeTaskOwner against the claiming TokenID), and the bridge — not the caller — assigns Seq and owns the per-step LogStepBytesMax budget, mirroring worker/log_writer.go so a consumer reading BUILD_LOGS sees the same chunk shape regardless of which lane produced it.
Subject/attempt/iteration scoping (#624 review rounds 2-3): the subject is logs.{runID}.{stepID}.{attempt}.{iteration}, matching worker/log_writer.go exactly — both are read from the claimed task’s own message (the same one authorizeTaskOwner already validated ownership against), NOT from the caller’s request body, so an HTTP worker can never spoof which attempt/iteration its chunks land on.
bridge/stranded_check.go Defense-in-depth counterpart to the engine’s primary stranded-subject check (Orchestrator.checkStrandedGroupSubjects, internal/engine): #704 moved the grouped subject/durable encoding behind the group sentinel (consumername.GroupSentinel) with no compatibility consumer, so a message published to a legacy grouped subject before every process upgrades is never reclaimed by any other mechanism -- see the pinned design on issue #704. This narrows the check to the one (task, group) pair the bridge is about to serve, checked right before adopting or creating that pair’s consumer.
Index
- func RegisterBridgeMetrics(m metric.Meter, b *Bridge) (metric.Registration, error)
- type AckMap
- func NewAckMap() *AckMap
- func (am *AckMap) Count() int64
- func (am *AckMap) Delete(taskID string)
- func (am *AckMap) Journal(taskID string) (*journal.Journal, bool)
- func (am *AckMap) JournalLock(taskID string) *sync.Mutex
- func (am *AckMap) JournalStop(taskID string) error
- func (am *AckMap) Load(taskID string) (jetstream.Msg, bool)
- func (am *AckMap) LoadWithTokenID(taskID string) (jetstream.Msg, string, bool)
- func (am *AckMap) MarkResolved(taskID, tokenID string)
- func (am *AckMap) PendingState(taskID string) ([]byte, bool)
- func (am *AckMap) SetJournal(taskID string, j *journal.Journal) bool
- func (am *AckMap) SetJournalStop(taskID string, err error) bool
- func (am *AckMap) SetPendingState(taskID string, state []byte) bool
- func (am *AckMap) Store(taskID string, msg jetstream.Msg, tokenID string)
- func (am *AckMap) Touch(taskID string) bool
- func (am *AckMap) WasResolvedBy(taskID string, claims workertoken.Claims) bool
- func (am *AckMap) WithLogState(taskID string, fn func(seq uint64, totalBytes int64, truncated bool) (newSeq uint64, newTotalBytes int64, newTruncated bool)) bool
- type Bridge
func RegisterBridgeMetrics
func RegisterBridgeMetrics(m metric.Meter, b *Bridge) (metric.Registration, error)RegisterBridgeMetrics wires the bridge’s observable instruments to m and returns the registration so a caller can unregister it.
The ackmap size is an observable gauge rather than an up/down counter deliberately. A counter requires every mutation site — store, resolve, reap, cap eviction — to Add the right delta forever, and drifts permanently the moment one is missed. This instrument was previously an Int64UpDownCounter with no Add call anywhere, so it reported a constant zero: indistinguishable from an idle bridge, and wrong the entire time it existed. A callback reading AckMap.Count() cannot drift, because it reports the real value at every collection.
Mirrors RegisterSchedulerMetrics (internal/trigger/metrics.go): a standalone registration function rather than construction inside NewBridge, so the error is returned to a caller that can assert on it instead of being discarded at startup.
type AckMap
AckMap tracks in-flight tasks for HTTP workers. Maps task_id ({runID}.{stepID}) to the NATS message so the bridge can ack/nak on behalf of the HTTP client when it resolves the task.
Thread-safe: multiple poll/resolve handlers run concurrently.
Bounded two ways, because an HTTP worker that dies mid-task never resolves and would otherwise leak its entry for the process lifetime: entries older than ackMapReapAfter are swept on insert, and ackMapMaxEntries caps the map between sweeps.
The sweep runs on insert rather than on a ticker because Bridge has no shutdown path to stop a goroutine against, and because entries are only ever created by traffic — an idle bridge cannot grow.
type AckMap struct {
// contains filtered or unexported fields
}func NewAckMap
func NewAckMap() *AckMapNewAckMap creates an empty AckMap ready for use.
func (*AckMap) Count
func (am *AckMap) Count() int64Count returns the number of in-flight tasks.
func (*AckMap) Delete
func (am *AckMap) Delete(taskID string)Delete removes a task from the map after resolution.
func (*AckMap) Journal
func (am *AckMap) Journal(taskID string) (*journal.Journal, bool)Journal returns the journal claim opened for taskID’s delivery.
func (*AckMap) JournalLock
func (am *AckMap) JournalLock(taskID string) *sync.MutexJournalLock returns the mutex that serializes journal actions on taskID’s claim, or nil when the task is not claimed.
func (*AckMap) JournalStop
func (am *AckMap) JournalStop(taskID string) errorJournalStop returns the fence recorded for taskID, or nil.
func (*AckMap) Load
func (am *AckMap) Load(taskID string) (jetstream.Msg, bool)Load retrieves the NATS message for the given task ID. Returns (nil, false) if not found. A thin wrapper over LoadWithTokenID for callers that don’t need the claiming TokenID.
Deliberately does not reap: a resolve arriving concurrently with the reaper must not race into a “task not found” that the worker cannot distinguish from a genuine unknown-task error.
func (*AckMap) LoadWithTokenID
func (am *AckMap) LoadWithTokenID(taskID string) (jetstream.Msg, string, bool)LoadWithTokenID retrieves the NATS message and the TokenID recorded at Store time for the given task ID (#627 per-task authorization: a resolve must present the same TokenID that claimed the task, or be Admin). Returns (nil, “”, false) if not found. Same does-not-reap contract as Load.
func (*AckMap) MarkResolved
func (am *AckMap) MarkResolved(taskID, tokenID string)MarkResolved records that taskID was just resolved (complete/fail/ pause/continue) by tokenID (empty for the admin bearer or dev mode) — called alongside Delete, from the same resolve call, so a POST /v1/tasks/{id}/logs that arrives afterwards can be told 409 (already resolved) instead of 404 (never existed) — but ONLY for the SAME caller that resolved it; see WasResolvedBy. Bounded and TTL’d the same way Store bounds entries: evicts the oldest resolvedTasks entry at resolvedTaskMax, and WasResolvedBy expires an entry past resolvedTaskTTL.
func (*AckMap) PendingState
func (am *AckMap) PendingState(taskID string) ([]byte, bool)PendingState returns the state buffered for taskID and whether a `checkpoint` set one since the claim.
func (*AckMap) SetJournal
func (am *AckMap) SetJournal(taskID string, j *journal.Journal) boolSetJournal records the journal claim for taskID and reports whether the entry exists. Never creates an entry.
func (*AckMap) SetJournalStop
func (am *AckMap) SetJournalStop(taskID string, err error) boolSetJournalStop records that taskID’s delivery was fenced, so every later action on it reports err. It reports whether the entry exists.
func (*AckMap) SetPendingState
func (am *AckMap) SetPendingState(taskID string, state []byte) boolSetPendingState buffers state for taskID’s next `continue` (#801) and reports whether the entry exists. Never creates an entry.
func (*AckMap) Store
func (am *AckMap) Store(taskID string, msg jetstream.Msg, tokenID string)Store saves a NATS message keyed by task ID, stamped with the insertion time and the TokenID of the caller that claimed it (#627; empty for the admin bearer or dev mode). Sweeps expired entries and enforces the size cap before inserting. Panics on empty taskID or nil msg — both are programmer errors.
func (*AckMap) Touch
func (am *AckMap) Touch(taskID string) boolTouch restarts taskID’s reap window and reports whether the entry exists. Call it after every successful msg.InProgress (#741): that call resets NATS’s AckWait, so the held delivery stays current and the reap window must count from it, not from the claim. Never creates an entry.
func (*AckMap) WasResolvedBy
func (am *AckMap) WasResolvedBy(taskID string, claims workertoken.Claims) boolWasResolvedBy reports whether taskID was resolved within the last resolvedTaskTTL BY claims — i.e. claims.Admin, or claims.TokenID equals the TokenID that resolved it (#624 review round 3: gating this on the caller, not just the taskID, closes an enumeration hole — without it, any valid token could probe 409-vs-404 for a task it never claimed and learn whether that task ever existed/resolved). Expires (and removes) a stale entry on read rather than waiting for a separate sweep — resolvedTasks is only ever queried from handleLogs’s rejection path, so a lazy expiry costs nothing extra there.
func (*AckMap) WithLogState
func (am *AckMap) WithLogState(taskID string, fn func(seq uint64, totalBytes int64, truncated bool) (newSeq uint64, newTotalBytes int64, newTruncated bool)) boolWithLogState atomically reads and updates the log-ingest counters for taskID under AckMap’s own mutex (#624). fn receives the current (seq, totalBytes, truncated) and returns the updated values, which are written back before the lock releases — so concurrent POST /v1/tasks/{id}/logs calls for the same task never race each other’s seq assignment or truncation decision. Returns false (fn not called) if taskID has no entry — same does-not-reap contract as LoadWithTokenID: a resolve racing the reaper must not surface as an ambiguous “not found” to the caller.
type Bridge
Bridge is an HTTP-to-NATS gateway that lets non-Go workers interact with DagNats over HTTP. Three deep endpoints expose the full worker lifecycle: connect, poll, and resolve.
Authentication (#627): the auth mode is decided by DAGNATS_BRIDGE_TOKEN ALONE -- no env token means open bridge (dev mode), regardless of whether a workertoken.Store is wired in via SetTokenStore. Set it and every request needs either the env token itself (admin/root, authenticates as workertoken.Claims{Admin: true}, bypassing task-type scoping entirely) or a minted, revocable, scoped worker token (“dgn_{id}_{secret}” bearer, checked against a configured Store). This is deliberate: minting a worker token itself requires the admin token (internal/api’s /v1/tokens routes fail closed with 503 when it is unset), so a Store-configured gate independent of the env token would let a fresh `dagnats serve` with no admin token lock out every worker permanently -- nothing could ever be minted to satisfy it. NewBridge logs a startup warning when the env token is unset so dev mode is never silent.
Every outbound NATS publish goes through *natsutil.TracingPublisher so W3C trace context (traceparent / tracestate) is auto-injected onto the outgoing message. This continues distributed traces from the inbound HTTP request into the NATS plane — without it, the trace ID would terminate at the HTTP boundary for non-Go workers.
type Bridge struct {
// contains filtered or unexported fields
}func NewBridge
func NewBridge(pub *natsutil.TracingPublisher) *BridgeNewBridge creates a Bridge. Panics on nil pub — a programmer error at startup. The TracingPublisher wraps both *nats.Conn and jetstream.JetStream and is the only legal publish surface inside this package (CI lint enforces this).
Binds optional KV buckets for checkpoints and signals (nil if not present).
func (*Bridge) Handler
func (b *Bridge) Handler() http.HandlerHandler returns an http.Handler with the three bridge routes. The mux routes are:
- POST /v1/workers/connect
- POST /v1/tasks/poll
- POST /v1/tasks/ (resolve, path includes task ID)
func (*Bridge) SetTokenStore
func (b *Bridge) SetTokenStore(store *workertoken.Store)SetTokenStore wires an optional workertoken.Store into the bridge so authMiddleware accepts minted worker tokens alongside the single env-configured admin bearer. nil is a valid value (clears it); the zero-value Bridge has no Store, matching pre-#627 behavior exactly.
Generated by gomarkdoc