@dokebi/nats

An optional plugin that gives a Dokebi application named, app-owned NATS connections. Version 0.4.0 (unpublished), used with @dokebi/core ^0.4.0. The real Docker integration checks of the source and the examples pass. AI agents start with AI.md.

Configuration and use

schemaVersion: 1
options:
  dokebi-nats:
    connections:
      primary:
        servers: ["nats://localhost:4222"]
      secondary:
        servers: ["nats://localhost:4223"]

The addresses are illustrations. Real addresses have no default; you state them. Independent servers get different connection names. The servers array of one connection lists candidates of the same NATS cluster.

| Key (per connection) | Type | Default | Description | |---|---|---|---| | servers | string[] | — (required) | DNS-only nats:// or tls:// candidates of one cluster | | connectTimeoutMs | integer ms | 5000 | First connect deadline | | shutdownTimeoutMs | integer ms | 5000 | Drain and close deadline | | healthIntervalMs | integer ms: 0, or 1000–3600000 | 10000 | PING period: a server that has not answered by the next PING is dropped as stale and reconnected, and the health check is down meanwhile (a frozen server is noticed within 2 × this); 0 keeps the SDK default (2 min, 2 unanswered) | | requestTimeoutMs | integer ms | 2000 | Request, JetStream ack and flush deadline | | reconnect | false | true | { maxAttempts, delayMs } | true (-1 = no limit, 2000) | Reconnect after a loss | | maxPending | integer 1–65536 | 4096 | Unresolved requests, JetStream publishes and consumer operations (0.4.1 raised it from 128 by the D-0102 rule: requests in flight per process × NATS operations each; about 4 KB of heap each plus the message) | | maxSubscriptions | integer | 128 | Subscriptions, consumers included | | subscriptionBufferMessages | integer | 4096 | Shared subscription buffer (messages). It must hold one socket read of small messages, which the SDK queues before any reader runs (0.4.1 raised it from 128: a burst of about 130 small events ended the subscription); about 540 bytes of heap per queued message besides its payload | | subscriptionBufferBytes | integer | 1048576 | Shared subscription buffer (bytes) | | maxPayloadBytes | integer | 1048576 | Message size, headers included | | maxPendingBytes | integer | 16777216 | Bytes sent but not yet flushed (at least maxPending × the usual message) | | busyRetryAfterMs | integer 1–60000 | 1000 | Retry-After of the retryable 503 NATS_BUSY | | unavailableRetryAfterMs | integer 1–60000 | 5000 | Retry-After of the retryable 503 NATS_UNAVAILABLE | | token, user, pass, tls | — | — | Authentication and TLS (below) | | health | enum | required | dokebi-health check nats:<name> |

0.4.0 renamed drainTimeoutMs → shutdownTimeoutMs, maxReconnectAttempts/reconnectTimeWaitMs → reconnect.maxAttempts/reconnect.delayMs, maxPendingRequests → maxPending and maxPendingPublishBytes → maxPendingBytes (D-0071). Old names still work in 0.4.0 with an OPTION_DEPRECATED warning. Error codes lost their NATS_ prefix in the same version (NATS_CLOSED → CLOSED).

import { Application, loadConfig } from "@dokebi/core";
import { createPlugin, dependency as nats } from "@dokebi/nats";

const document = await loadConfig("/absolute/path/dokebi.yaml");
const app = new Application({ config: document.options });
app.register(createPlugin({ app: app.view }));
app.register({
  metadata: {
    name: "example-consumer",
    version: "1.0.0",
    core: "^0.4.0",
    dependencies: [nats],
  },
  setup(context) {
    const provider = context.getDependency(nats);
    const primary = provider.connection("primary");
    // Build the services that use primary, and their cleanup, here.
    return primary;
  },
});
await app.start();
await app.stop();

The factory input is { app, options? }; app (the app.view core created) is required (plugin contract v2). Without options it reads its own app.config['dokebi-nats']; given options are used alone. The two are never merged. Invalid options fail with INVALID_PLUGIN_OPTIONS (cause: the core INVALID_OPTIONS, with a list of paths and problems in details.issues and no values; since 0.4.0 one defineOptions declaration in src/options.ts builds both the schema and the parser, D-0063). Invalid input (dependencies and similar) is INVALID_PLUGIN_INPUT. An instance can be set up once, only in the app that created it (PLUGIN_ALREADY_STARTED, PLUGIN_APP_MISMATCH). Internal normalized defaults never change app.config.

dependency is the provider descriptor for user code (also at @dokebi/nats/dependency). A consumer that puts it in metadata.dependencies reuses a registered dokebi-nats or gets one created from this package. When you register it yourself, as above, there is only one provider.

Logs

Each connection state change is logged once by a child logger carrying the connection name, with code as the only field: connected NATS_CONNECTED (info), disconnected NATS_DISCONNECTED (warn), recovered NATS_RECONNECTED (info), and an abnormal close outside drain with the close error code (error; the SDK's reason in full in the error field when there is one). A repeated state is not logged again. Log codes keep the NATS_ prefix.

Connection API

provider.names lists the connection names and provider.connection(name) returns a connection handle. The handle has name, state (connected, reconnecting, draining, closed), closed: Promise<NatsError | undefined> and the operations below. closed resolves to undefined on a normal close and to a sanitized error on failure.

• publish(subject, data) sends bytes and confirms the server flush. It does not mean the message is stored or processed.
• request(subject, data) returns the reply { subject, data, headers?, reply? }.
• subscribe(subject, { queue? }) is an async iterator of messages. Confirm await subscription.ready before another connection publishes. Watch closed for the outcome and end it with unsubscribe(). A message offers respond(data).
• jetstream.publish(subject, data, { messageId?, headers? }) returns the storage confirmation { stream, seq, duplicate, domain? }. The target stream must be created separately.

The raw SDK client and the right to close the connection are not exposed. The runtime offers Core messaging, JetStream publish and limited handling of durable consumers created beforehand. Management runs explicitly in the separate provision entry, and the runtime never changes resources on its own. Handlers, retries and failed-job policy belong to job.

Limits and shutdown

All connections are prepared in setup, with cleanup registered first. When only some connections are ready and setup fails, the acquired ones are cleaned up. Consumers borrow connections and the provider drains them in reverse-order cleanup. Work that finishes before the next next(), such as await message.respond(...) inside a normal for await body, is tracked by drain. Detached async work is not. unsubscribe() and the iterator's return() end consumption and drop waiting messages. A consumer cleanup that waits only for a loop with no end condition may never let the provider drain.

During a reconnect new sends are rejected, and application work is never resent automatically. A disconnect during an unresolved flush or subscription setup fails it with FLUSH_FAILED. A failed flush may leave the outcome of work already sent unknown. A request slot returns when the SDK promise ends, sent bytes when the flush ends. The subscription buffer is shared by all subscriptions of the connection; going over the limit fails that subscription.

Start and shutdown have deadlines. When an SDK connect never returns and cleanup cannot be confirmed, CLEANUP_TIMEOUT is reported. When the connection returns late it tries to close it, but resource release by an SDK that hangs forever is not promised. Errors never expose the original server reply, addresses, credentials, file paths or the original cause.

Authentication and TLS

Set token or a user/pass pair, never both. tls: {} requires TLS; caFile, certFile and keyFile take absolute paths. A client certificate and key come together. There is no option to turn off certificate verification. The real Docker regression of authentication and TLS passes.

Runnable example

After building the repository, node examples/11-nats/run.mjs prepares two independent NATS servers and a JetStream stream in its own Docker Compose project, runs the example with a temporary YAML, then cleans up. To use servers you already run: node examples/11-nats/main.mjs /absolute/path/dokebi.yaml. The stream management rights of the example's setup script are not part of the provider's consumer API. The automatic run of example 11 passes. The earlier authentication/TLS and fault checks pass and are kept apart from the newer consumer checks.

Check the licenses of the Dokebi source and of the bundled SDK. Both NATS SDK packages are Apache-2.0. Read the AI guide as well.

SDK address restriction

To block the IP TLS host verification defect of SDK 3.4.0, every servers candidate must be a DNS name. A plaintext URL can still switch to TLS automatically, so IP addresses and mixed DNS/IP lists are configuration errors (INVALID_OPTIONS in provisionQueueResources). Use localhost explicitly for local development. Addresses are never rewritten. The restriction will be reconsidered once a fixed official version passes real IP SAN, DNS, multi-candidate and reconnect regressions.

JetStream consumer

connection.jetstream.bindConsumer(options) binds to a stream and a named durable pull consumer created beforehand. Give all of stream, consumer, subject, maxInFlight, maxMessageBytes and ackWaitMs. It checks a file/WorkQueue stream, the exact subject, explicit acknowledgement, the ack wait and the pending limits, and returns a fixed error on a mismatch. It never creates or changes resources.

consumer.next(signal?) returns a delivery, or null for an empty pull. close() releases the consumer. A delivery offers data, streamSequence, deliveryCount, headers (0.4.1, D-0108: the stored message's headers in the same immutable form as received Core messages, absent when it has none; job reads the attempts a handed-back job carries), ack(), retry(delayMs), working() and release(). ack waits for the server's confirmation; retry and working wait for the shared flush. release gives up local ownership and is not a success acknowledgement. Choose ack or retry from the result and release in finally. Heartbeats and running handlers belong to the higher processing layer.

Consumer bindings share maxSubscriptions with subscriptions. Work in flight uses maxPending and the send byte limit (maxPendingBytes). A public timeout alone never returns an SDK slot early, and a delivery that arrives late is released without being handed out. Shutdown cancels pulls in flight and releases the consumer and SDK resources within the same bounded deadline.

Explicit provisioning

Use provisionQueueResources({ connection, resources }) from the separate @dokebi/nats/provision. connection follows the provider's DNS, authentication and TLS normalization, including the DNS-only rule.

resources spells out all of stream, consumer, subject, deadLetterStream, deadLetterSubject, maxPayloadBytes, ackWaitMs, maxAckPending, maxBytes, maxMessages, duplicateWindowMs and replicas. It creates a missing main WorkQueue stream, a Limits stream for failed jobs and the durable consumer, or compares them with the existing contract. Both streams use file storage and reject new publishes when full. Existing resources are never edited or deleted, and a failure after a partial creation is not rolled back. Running the same declaration again creates what is left and compares the rest.

Success returns the created and existing lists; failure is a sanitized NatsProvisionError (INVALID_OPTIONS, PROVISION_MISMATCH, PROVISION_FAILED, PROVISION_TIMEOUT, PROVISION_CLEANUP). A connection that opens after the connect deadline is closed too. Unconfirmed cleanup is reported as cleanupUnconfirmed and never treated as success. When the JetStream server refused a call, server is its answer { code, description } (also in the message); connection errors are never included. job offers the dokebi-job-provision command that uses it.

Verification boundary of recent changes

The original connection features passed 28 unit, 14 security, 6 basic Docker integration and 5 fault tests and a standalone CLI install. Those numbers predate the consumer. The 11 new consumer/connection regressions, job example 16 and scheduler example 17 pass too. Example 16 was checked again after the last consumer cleanup change. The real TypeScript declaration graph confirms no public SDK types leak. The earlier security matrix was not repeated.

Publish headers and pre-check

jetstream.publish headers are a read-only string map and expose no SDK type. Headers of received messages and replies are a map of immutable string arrays that keep duplicate values. User headers are limited to 32 and 16384 bytes in total; control characters, duplicate names, reserved Nats-* names and accessors or proxies are rejected. The real wire headers, message ID included, and the body count toward the provider and server limits.

jetstream.checkPublish({ stream, subject, maxPayloadBytes, messageId?, headers? }) checks the target stream and the largest send size without changing resources. Too large is MAX_PAYLOAD, a target or contract mismatch PUBLISH_MISMATCH. The management lookup uses the connection's request limit and deadline. It cannot promise success against server changes or a full store after the lookup.

maxHeadersBytes of bindConsumer is optional (default 16384, range 0–16384) and separate from the body's maxMessageBytes. Binding requires the stream's max_msg_size to be at most maxMessageBytes + maxHeadersBytes (and the provider and server limits); receiving then delivers every stored message, also a body over maxMessageBytes that came with fewer headers (NATS counts both together), so the consumer handles it as poison instead of the pull failing on every redelivery. Job uses 155 bytes for the source message ID. provisionQueueResources builds stream limits that include the source/DLQ header budget and rejects insufficient provider or server limits before creating anything.

First connect retry (2026-10-14)

With reconnect on (default), a failed first connect is retried within connectTimeoutMs from 100 ms, doubling up to reconnect.delayMs. An app that starts during a short NATS restart does not fail. Past the deadline it fails as before. With reconnect: false it tries once (ISSUE-0005 PG-23).