Documentation
¶
Overview ¶
Package worker is the shared runtime for TMI Component Platform worker binaries (tmi-extractor, tmi-chunk-embed). It owns the NATS JetStream connection bootstrap, the durable-consumer loop with at-least-once and idempotency handling, and the heartbeat publisher. It is framework-free: no Gin, no GORM, no internal/config — a worker binary that imports this package and pkg/extract / pkg/jobenvelope pulls in nothing else from the monolith.
Index ¶
- Constants
- func ChunkEmbedSubject(jobID string) string
- func ConsumerNameFor(componentName string) string
- func EnvDuration(key string, fallback time.Duration) time.Duration
- func EnvOr(key, fallback string) string
- func HeartbeatSubject(component string) string
- func MustEnv(key string) (string, error)
- func PayloadRefForJob(ref, jobID string) bool
- func ResultSubject(jobID string) string
- func RunConsumer(ctx context.Context, conn *Conn, cfg ConsumerConfig, handle JobHandler) error
- func RunHeartbeat(ctx context.Context, conn *Conn, instanceID string, interval time.Duration)
- func SanitizeName(s string) string
- func StreamNameFor(componentName string) string
- type Config
- type Conn
- func (c *Conn) Close()
- func (c *Conn) Config() Config
- func (c *Conn) DeletePayload(ctx context.Context, ref string) error
- func (c *Conn) GetPayload(ctx context.Context, ref string) ([]byte, error)
- func (c *Conn) JetStream() jetstream.JetStream
- func (c *Conn) Publish(ctx context.Context, subject string, data []byte) error
- func (c *Conn) PublishCore(subject string, data []byte) error
- func (c *Conn) PublishFailureResult(ctx context.Context, jobID, reasonCode, detail string) error
- func (c *Conn) PutPayload(ctx context.Context, name string, data []byte) (string, error)
- type ConsumerConfig
- type Heartbeat
- type JobError
- type JobHandler
Constants ¶
const ( // OutcomeAck marks the message processed successfully. OutcomeAck outcome = iota // OutcomeNak requests redelivery (transient failure). OutcomeNak // OutcomeTerm permanently drops the message (terminal failure). OutcomeTerm )
const ( // SubjectExtractPrefix is the NATS subject prefix for extraction job messages // (e.g., "jobs.extract.<jobID>"). SubjectExtractPrefix = "jobs.extract." // SubjectChunkEmbedPrefix is the NATS subject prefix for chunk-and-embed job // messages (e.g., "jobs.chunkembed.<jobID>"). SubjectChunkEmbedPrefix = "jobs.chunkembed." // SubjectResultPrefix is the NATS subject prefix for job result messages // (e.g., "jobs.result.<jobID>"). SubjectResultPrefix = "jobs.result." // SubjectDLQ is the NATS subject for dead-letter-queue messages: jobs that // have exhausted all delivery retries. SubjectDLQ = "jobs.dlq" // SubjectMaxDeliverAdvisory is the wildcard subject on which JetStream // publishes a MAX_DELIVERIES advisory when a message exhausts a consumer's // MaxDeliver. The concrete subject is // "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.<stream>.<consumer>". SubjectMaxDeliverAdvisory = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>" // SubjectHeartbeatPrefix is the NATS subject prefix for component heartbeat // messages (e.g., "components.heartbeat.<componentName>"). SubjectHeartbeatPrefix = "components.heartbeat." )
const DLQAdvisoryStream = "TMI_DLQ_ADVISORY"
DLQAdvisoryStream is the JetStream stream that durably captures MAX_DELIVERIES consumer advisories so the monolith's DLQ producer survives restarts (a plain core-NATS subscription would miss advisories fired while the monolith is down).
const DLQStream = "TMI_DLQ"
DLQStream is the dead-letter JetStream stream bound to SubjectDLQ ("jobs.dlq"). The monolith creates it, publishes dead-lettered Job envelopes to it, and consumes from it (see api/dlq_producer.go and the ResultConsumer DLQ subscription). It is not owned by any per-component stream.
const JobsStreamPrefix = "TMI_"
JobsStreamPrefix is the prefix the controller's streamNameFor prepends to the sanitized component name to produce the per-component JetStream stream name. Plan 1's render_jetstream.go is the source of truth for this value.
const PayloadBucket = "TMI_PAYLOADS"
PayloadBucket is the JetStream Object Store bucket name used for payload-by-reference: large job payloads are stored here and the job envelope carries only the object key. Plan 2 creates this bucket at worker startup; the Plan 1 controller does not render it.
const ResultStream = "TMI_RESULTS"
ResultStream is the dedicated JetStream stream that Plan 2 workers create for jobs.result.* subjects. It is not owned by any Plan 1 per-component stream; the monolith result-consumer (Plan 3) and the workers both bind it.
Variables ¶
This section is empty.
Functions ¶
func ChunkEmbedSubject ¶
ChunkEmbedSubject returns the NATS subject for a chunk-and-embed job with the given job ID.
func ConsumerNameFor ¶
ConsumerNameFor returns the durable JetStream consumer name for a component, mirroring the controller's consumerNameFor function in render_jetstream.go. Example: "tmi-chunk-embed" → "TMI_CHUNK_EMBED_CONSUMER".
func EnvDuration ¶
EnvDuration parses key as a Go duration, returning fallback if it is unset/empty or fails to parse.
func HeartbeatSubject ¶
HeartbeatSubject returns the NATS subject for a component's heartbeat messages. The component argument is passed through UNSANITIZED: NATS subjects use "." as a hierarchy delimiter (not an illegal character), so sanitizing would corrupt the subject hierarchy.
func PayloadRefForJob ¶
PayloadRefForJob reports whether an object_ref names a blob that belongs to the given job. Blob names embed the job ID in one of the shipped patterns: the publisher's input blob ("job-<id>-source") or a stage-output blob whose leading path segment is the job ID ("<id>/extracted", "<id>/result"). Envelope-supplied refs are worker-controlled, so consumers MUST check this before acting on a ref — honoring an arbitrary ref would let a forged envelope delete another job's blobs.
func ResultSubject ¶
ResultSubject returns the NATS subject for the result of a specific job.
func RunConsumer ¶
func RunConsumer(ctx context.Context, conn *Conn, cfg ConsumerConfig, handle JobHandler) error
RunConsumer creates the durable consumer and dispatches messages to the handler until ctx is cancelled. It blocks; run it on the main goroutine.
func RunHeartbeat ¶
RunHeartbeat publishes a Heartbeat on the component's heartbeat subject every interval until ctx is cancelled. It is meant to run in its own goroutine; a publish failure is logged and retried on the next tick.
func SanitizeName ¶
SanitizeName upcases s and replaces JetStream-illegal characters (".", "-", " ") with "_". It mirrors the unexported sanitizeName function in internal/platform/controller/render_jetstream.go; the two implementations MUST stay in sync so that a worker can derive the same stream and consumer names as the controller.
func StreamNameFor ¶
StreamNameFor returns the JetStream stream name for a component, mirroring the controller's streamNameFor function in render_jetstream.go. Example: "tmi-extractor" → "TMI_TMI_EXTRACTOR".
Types ¶
type Config ¶
type Config struct {
// NATSURL is the NATS server URL (env TMI_NATS_URL).
NATSURL string
// ComponentName is this worker's TMIComponent name (env TMI_COMPONENT_NAME).
ComponentName string
}
Config is the worker's NATS bootstrap configuration, read from env vars.
func ConfigFromEnv ¶
ConfigFromEnv builds a Config from the standard worker env vars.
type Conn ¶
type Conn struct {
// contains filtered or unexported fields
}
Conn bundles a NATS connection, a JetStream context, and the payload Object Store handle. It is the worker's single handle to the bus.
func Connect ¶
Connect dials NATS, opens a JetStream context, and ensures the payload Object Store bucket exists.
If TMI_NATS_CREDS is set in the environment, it is used as the path to a NATS credentials file, giving each component its own bus identity once the server enables authorization. Unset preserves the credential-less default. If TMI_NATS_TLS_CA_FILE is set (see internal/tlsconfig), the connection uses TLS with that CA and, when TMI_NATS_TLS_CERT_FILE/TMI_NATS_TLS_KEY_FILE are set, a client certificate.
func (*Conn) Close ¶
func (c *Conn) Close()
Close closes the NATS connection. In-flight messages are not drained.
func (*Conn) DeletePayload ¶
DeletePayload removes a blob by the object_ref produced by PutPayload. It is idempotent from the caller's perspective: deleting an absent blob is treated as success so result-consumer cleanup never blocks on a double-delivery.
func (*Conn) GetPayload ¶
GetPayload reads a blob by the object_ref produced by PutPayload.
func (*Conn) Publish ¶
Publish publishes a pre-marshaled message to a JetStream subject (durable, stream-backed). For ephemeral signals such as heartbeats use PublishCore.
func (*Conn) PublishCore ¶
PublishCore publishes a message over core NATS (fire-and-forget, no JetStream stream or persistence). Use this for ephemeral signals such as heartbeats. Job messages that must be durable go through Publish.
func (*Conn) PublishFailureResult ¶
PublishFailureResult publishes a terminal "failed" Result envelope for the given job and returns a terminal *JobError so the consumer terminates the message. Handlers perform their own error classification to derive reasonCode/detail, then call this shared helper for the common build-Result -> marshal -> publish -> return-terminal-JobError core. A marshal or publish failure is returned as-is (the latter is transient and triggers redelivery).
type ConsumerConfig ¶
type ConsumerConfig struct {
// StreamName is the JetStream stream the consumer binds. The controller
// renders one stream per component (TMI_<NAME>); RunConsumer creates the
// stream if it does not yet exist.
StreamName string
// Durable is the durable consumer name (stable across restarts).
Durable string
// FilterSubject is the subject filter (e.g. "jobs.extract.>").
FilterSubject string
// AckWait is the redelivery timeout — also the JetStream-side backstop
// for a worker that dies mid-job.
AckWait time.Duration
// MaxDeliver caps redeliveries before JetStream dead-letters.
MaxDeliver int
}
ConsumerConfig configures the durable consumer.
type Heartbeat ¶
type Heartbeat struct {
// Component is the TMIComponent name.
Component string `json:"component"`
// InstanceID identifies the publishing pod/process.
InstanceID string `json:"instance_id"`
// SentAt is the publish timestamp.
SentAt time.Time `json:"sent_at"`
}
Heartbeat is the liveness message a worker publishes on components.heartbeat.<component>. The monolith uses it to distinguish "type declared, no healthy instance" from "instances present".
type JobError ¶
type JobError struct {
// ReasonCode is a pkg/extract Reason* constant (or a status string).
ReasonCode string
// Detail is optional human-readable context.
Detail string
// Terminal is true when redelivery cannot help.
Terminal bool
}
JobError is a typed, terminal failure a handler returns when a job can never succeed (malformed input, unsupported format, timeout). The consumer terminates such a message rather than redelivering it.
type JobHandler ¶
type JobHandler func(ctx context.Context, job jobenvelope.Job) error
JobHandler processes one decoded job. Returning nil acks the message; returning a terminal *JobError terminates it; any other error naks it for redelivery. The handler is responsible for publishing the result envelope for terminal failures BEFORE returning the *JobError — the consumer only decides ack/nak/term, it does not publish results.
Source Files