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. SEM@6df2ea658f2595a125e17591a26dbca6df6f71ad: build the NATS subject for dispatching a chunk-embed job by job ID (pure)
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". SEM@6df2ea658f2595a125e17591a26dbca6df6f71ad: build the durable JetStream consumer name for a worker component, matching the KEDA ScaledObject (pure)
func EnvDuration ¶
EnvDuration parses key as a Go duration, returning fallback if it is unset/empty or fails to parse. SEM@fb53ed13960634e8e70bfe57a75d2305f28c4174: parse an environment variable as a duration, returning a fallback on missing or invalid input (pure)
func EnvOr ¶
EnvOr returns the value of key, or fallback if it is unset/empty. SEM@fb53ed13960634e8e70bfe57a75d2305f28c4174: fetch an environment variable with a fallback default (pure)
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. SEM@6df2ea658f2595a125e17591a26dbca6df6f71ad: build the NATS subject for a worker component heartbeat by component name (pure)
func MustEnv ¶
MustEnv returns the value of key or an error if it is unset/empty. SEM@fb53ed13960634e8e70bfe57a75d2305f28c4174: fetch a required environment variable or return an error if absent (pure)
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. SEM@d056a3ea026249d40d05ab6af7f092a043f72c7a: validate that an object ref belongs to the given job ID, rejecting forged refs (pure)
func ResultSubject ¶
ResultSubject returns the NATS subject for the result of a specific job. SEM@6df2ea658f2595a125e17591a26dbca6df6f71ad: build the NATS subject for publishing a job result by job ID (pure)
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. SEM@e69b1723153a31aa74eb58c885a3ca54a9cbb016: bind a durable JetStream consumer and dispatch jobs to the handler until context is cancelled
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. SEM@71c1d8554ecca870da2bafa898b79d1c29d43ebf: publish periodic heartbeat messages to NATS until context is cancelled (mutates shared state)
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. SEM@6df2ea658f2595a125e17591a26dbca6df6f71ad: convert a component name to an uppercase, underscore-safe NATS identifier (pure)
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". SEM@6df2ea658f2595a125e17591a26dbca6df6f71ad: build the JetStream stream name for a worker component (pure)
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. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: NATS bootstrap configuration holding server URL and component name (pure)
func ConfigFromEnv ¶
ConfigFromEnv builds a Config from the standard worker env vars. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: build NATS worker config from required environment variables (pure)
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. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: worker handle bundling a NATS connection, JetStream context, and payload object store (pure)
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. SEM@946ec29: connect to NATS (TLS when configured), open JetStream, and ensure the payload object store
func (*Conn) Close ¶
func (c *Conn) Close()
Close closes the NATS connection. In-flight messages are not drained. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: close the NATS connection without draining in-flight messages
func (*Conn) Config ¶
Config returns the connection's config. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: return the connection's NATS bootstrap config (pure)
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. SEM@b4d93cd08868f7468a93b099aca5217f7b263d56: delete a payload blob by object ref; idempotent if already absent
func (*Conn) GetPayload ¶
GetPayload reads a blob by the object_ref produced by PutPayload. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: fetch a blob from the payload object store by object ref
func (*Conn) JetStream ¶
JetStream returns the JetStream context for consumer/publish wiring. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: return the JetStream context for consumer and publish wiring (pure)
func (*Conn) Publish ¶
Publish publishes a pre-marshaled message to a JetStream subject (durable, stream-backed). For ephemeral signals such as heartbeats use PublishCore. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: publish a message to a durable JetStream subject
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. SEM@71c1d8554ecca870da2bafa898b79d1c29d43ebf: publish a fire-and-forget message over core NATS without persistence
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). SEM@fcd7743e746718c31b33ef56fb3ba2f8ccf669c7: publish a terminal job failure result and return a terminal job error (mutates shared state)
func (*Conn) PutPayload ¶
PutPayload writes bytes to the Object Store under the given name and returns the object_ref to carry in an envelope. SEM@9d9b68ad12f48fd7a991a43f29615cea731c76ca: store a blob in the payload object store and return its object ref
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. SEM@3b4afc57df700de14d06ec4e93a7038dcf52b9d2: parameters for binding a durable JetStream consumer to a stream and subject (pure)
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". SEM@0e54997b95671dd5ea0319130f237e438d611479: heartbeat payload carrying component name, instance ID, and publish timestamp (pure)
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. SEM@3b4afc57df700de14d06ec4e93a7038dcf52b9d2: typed terminal job failure signalling whether redelivery can help (pure)
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. SEM@3b4afc57df700de14d06ec4e93a7038dcf52b9d2: callback contract for processing a decoded job and signalling ack/nak/term via error (pure)
Source Files