# Instructions to use NATS

## NATS Grid Summary - Timing Statistics Deep Analysis

### The Mystery: Why "Network receive time" < "Receive time"?

**Observed confusing output:**

```text
## NATS Summary
Send time         │   0.01 s
Receive time      │   4.45 s

## NATS Grid Summary
Network send time    │           3.05 sec
Network receive time │           0.55 sec
```

**The question:** How can "Network receive time" (0.55s) be smaller than "Receive time" (4.45s)?

**The answer:** These are measuring DIFFERENT things at DIFFERENT abstraction levels.

---

### The Two Timing Systems

#### System 1: NATS Connection Level (nats.lua)

**Location:** `/Volumes/nc/nc-server/lib/message/nats/nats.lua`

**What it measures:** Raw TCP socket operations at the protocol level.

**Send time** (lines 629-631):

```lua
local sendStart = util.seconds()
self:socketWrite(protoCmd .. headerBytes .. msg.data .. api.CRLF)
self.stats.send_time = self.stats.send_time + util.seconds(sendStart)
```

- Measures: Time to write to TCP socket
- Accumulates: In `publishMsg()` - every publish call
- Scope: Protocol-level write operation only

**Receive time** (lines 325-327):

```lua
local recvStart = util.seconds()
readData, sockErr, readLen = self.sock:receive(api.CRLF, timeout)
self.stats.receive_time = self.stats.receive_time + util.seconds(recvStart)
```

- Measures: Time to read from TCP socket
- Accumulates: In `processIncoming()` - every socket read call
- Scope: Protocol-level read operation only

**Key insight:** These are LOW-LEVEL TCP socket timings, accumulated across ALL operations (publish, subscribe, responses, heartbeats, etc.).

---

#### System 2: NATS Grid Level (nats-grid.lua)

**Location:** `/Volumes/nc/nc-server/lib/message/nats/nats-grid.lua`

**What it measures:** High-level application operations with data processing overhead.

**Send operation timing** (lines 272-444):

```lua
function grid.sendGrid(bucket, gridData, option)
    local startTime = util.seconds()  -- Line 273: WALL-CLOCK START

    -- For each chunk:
    --   1. JSON serialization (accumulated to totalToJsonTime)
    --   2. Compression (accumulated to totalCompressionTime)
    --   3. Network send (accumulated to totalNetworkSendTime):

    local networkSendTimer = util.seconds()  -- Line 397
    ret = js:publish(subject, msgJson)       -- Line 398
    local networkSendTime = util.seconds(networkSendTimer)  -- Line 403
    totalNetworkSendTime = totalNetworkSendTime + networkSendTime

    totalSendTime = totalSendTime + util.seconds(startTime)  -- Line 444: WALL-CLOCK END
end
```

**Receive operation timing** (lines 532-636):

```lua
function grid.receiveGrid(bucket, option)
    local startTime = util.seconds()  -- Line 533: WALL-CLOCK START

    -- Network receive:
    local networkReceiveTime = util.seconds()  -- Line 538
    local consumer = grid.createOrderedConsumer(bucket, option)
    local msgArray = consumer:fetch(option.message_count, fetchOpt)  -- Line 553
    networkReceiveTime = util.seconds(networkReceiveTime)  -- Line 557

    -- Data processing (lines 576-613):
    for i, msg in ipairs(msgArray) do
        -- Decompression (accumulated to totalDecompressionTime)
        -- JSON parsing (accumulated to totalFromJsonTime)
    end

    totalNetworkReceiveTime = totalNetworkReceiveTime + networkReceiveTime  -- Line 625
    totalReceiveTime = totalReceiveTime + util.seconds(startTime)  -- Line 635: WALL-CLOCK END
end
```

**Key insight:** Grid level measures WALL-CLOCK time including ALL processing overhead.

---

### The Timing Hierarchy

```text
┌─────────────────────────────────────────────────────────────────────────────┐
│                    NATS CONNECTION LEVEL (nats.lua)                        │
├─────────────────────────────────────────────────────────────────────────────┤
│ send_time    = Accumulated socket:write() time only                        │
│ receive_time = Accumulated socket:receive() time only                      │
│                                                                              │
│ Measures: Raw TCP protocol operations                                       │
│ Scope: ALL operations (pub, sub, responses, heartbeats, flow control)      │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                    NATS GRID LEVEL (nats-grid.lua)                         │
├─────────────────────────────────────────────────────────────────────────────┤
│ SEND OPERATION:                                                             │
│   totalSendTime = startTime → endTime (wall-clock)                         │
│     = ToJsonTime + CompressionTime + NetworkSendTime                       │
│                                                                              │
│ RECEIVE OPERATION:                                                          │
│   totalReceiveTime = startTime → endTime (wall-clock)                      │
│     = NetworkReceiveTime + DecompressTime + FromJsonTime                   │
│                                                                              │
│ Measures: Application-level operations with data processing                 │
│ Scope: Only sendGrid() and receiveGrid() operations                        │
└─────────────────────────────────────────────────────────────────────────────┘
```

---

### Why Your Test Shows These Values

**Test output analysis:**

```text
## NATS Grid Summary
Total time           │          14.85 sec
----------------------------------------------------
Network send time    │           3.05 sec    ← js:publish() calls only
Compress time        │         8.5754 sec    ← zstd compression
To JSON time         │         0.6163 sec    ← toJsonRaw()
Total send time      │          12.49 sec    ← Wall-clock sendGrid()
                              3.05 + 8.58 + 0.62 ≈ 12.25 sec ✓
----------------------------------------------------
Network receive time │           0.55 sec    ← consumer:fetch() only
Decompress time      │         0.1533 sec    ← zstd decompression
From JSON time       │         1.6543 sec    ← json.fromJson()
Total receive time   │           2.36 sec    ← Wall-clock receiveGrid()
                              0.55 + 0.15 + 1.65 ≈ 2.35 sec ✓
```

**Note:** The slight difference between sum and total is due to consumer creation overhead and timing precision.

---

### The Critical Finding: Send/Receive Overlap During Request/Response

**This is where it gets interesting:**

During `sendGrid()`, the code uses `js:publish()` which is a REQUEST/RESPONSE pattern:

```lua
-- nats-jetstream.lua:346
self.connection:RequestMsgWithContext(message)
  └── nats.lua:692
      ├── publishMsg() → Sends to server
      └── waitForResponse() → Loops calling processIncoming()
          └── processIncoming() → Reads socket, handles incoming messages
```

**The response subscription is ALWAYS active:**

```lua
-- nats.lua:583 - Created once during connection
self.responseSubscription = self:subscribe("_INBOX.*", api.NO_QUEUE_GROUP)
```

**What happens during send:**

1. Client publishes with reply subject `_INBOX.<token>`
2. Server processes and publishes response to `_INBOX.<token>`
3. `waitForResponse()` loops calling `processIncoming()`
4. `processIncoming()` reads socket (accumulating `receive_time` in nats.lua!)
5. Response arrives, `processMsg()` sets `responseMap[token].done = true`

**THE KEY INSIGHT:**

`nats.lua` `receive_time` accumulates during SEND operations because responses arrive on the socket while waiting for `js:publish()` to complete!

---

### Why NATS Summary Shows Different Values

**From test output:**

```text
## NATS Summary
Send time         │   0.01 s
Receive time      │   4.45 s
```

**Explanation:**

- `Send time` (0.01s) = Accumulated `socket:write()` time in nats.lua
  - Very fast because socket writes are buffered and return immediately
  - Does NOT include time waiting for responses

- `Receive time` (4.45s) = Accumulated `socket:receive()` time in nats.lua
  - Includes ALL socket reads during:
    - `js:publish()` waiting for JetStream ACK (during sendGrid!)
    - `consumer:fetch()` receiving messages (during receiveGrid!)
    - Any heartbeats, flow control, INFO updates

**Why 4.45s is larger than 0.55s (Network receive time):**

| Metric | Value | What it includes |
|--------|-------|------------------|
| nats.lua `receive_time` | 4.45s | ALL socket reads (send responses + fetch + heartbeats) |
| nats-grid.lua `networkReceiveTime` | 0.55s | Only `consumer:fetch()` call duration |

---

### The Complete Timing Breakdown

```text
sendGrid() operation:
┌────────────────────────────────────────────────────────────────┐
│ totalSendTime = 12.49 sec                                       │
│                                                                 │
│ Breakdown:                                                      │
│   toJsonRaw()         → 0.62 sec                               │
│   zstd.compress()     → 8.58 sec                               │
│   js:publish()        → 3.05 sec (totalNetworkSendTime)        │
│       ├─ socket:write()    → ~0.01s (accumulates send_time)    │
│       ├─ waitForResponse() → reads socket for ACK              │
│       │   └─ socket:receive() → accumulates receive_time!      │
│       └─ processMsg()      → handles ACK response              │
└────────────────────────────────────────────────────────────────┘

receiveGrid() operation:
┌────────────────────────────────────────────────────────────────┐
│ totalReceiveTime = 2.36 sec                                     │
│                                                                 │
│ Breakdown:                                                      │
│   createOrderedConsumer() → minimal overhead                    │
│   consumer:fetch()      → 0.55 sec (totalNetworkReceiveTime)   │
│       └─ socket:receive() → accumulates receive_time!          │
│   zstd.decompress()     → 0.15 sec                             │
│   json.fromJson()       → 1.65 sec                             │
└────────────────────────────────────────────────────────────────┘
```

---

### Answer to the Original Question

**Q: How can "Network receive time" (0.55s) be smaller than "Receive time" (4.45s)?**

**A:** They measure different things:

1. **"Network receive time"** (0.55s) = `consumer:fetch()` call duration only
   - Pure JetStream fetch operation
   - Measured at nats-grid.lua application level

2. **"Receive time"** (4.45s) = Accumulated ALL socket receive time
   - Includes socket reads during `js:publish()` (waiting for ACK during SEND!)
   - Includes socket reads during `consumer:fetch()`
   - Includes any heartbeat, flow control messages
   - Measured at nats.lua TCP protocol level

**The formula:**

```text
nats.lua receive_time (4.45s)
  = socket reads during sendGrid() (js:publish() responses)
  + socket reads during receiveGrid() (consumer:fetch())
  + socket reads for heartbeats/flow control

nats-grid.lua totalNetworkReceiveTime (0.55s)
  = consumer:fetch() call duration only
```

**Even without coroutines enabled**, the request/response pattern means the socket is being read during send operations, accumulating `receive_time` in nats.lua stats.

---

### Summary: Which Metric to Use?

| Question | Use Metric |
|----------|------------|
| How long did sendGrid() take? | `totalSendTime` (Grid Summary) |
| How long did receiveGrid() take? | `totalReceiveTime` (Grid Summary) |
| Pure network send latency | `networkSendTime` (Grid Summary) |
| Pure network receive latency | `networkReceiveTime` (Grid Summary) |
| Compression overhead | `totalCompressionTime` |
| JSON serialization overhead | `totalToJsonTime` / `totalFromJsonTime` |
| Raw TCP socket write time | `send_time` (NATS Summary) |
| Raw TCP socket read time | `receive_time` (NATS Summary) - **includes reads during send!** |

---

### Code References

- **nats.lua:629-631** - `send_time` measurement (socket write)
- **nats.lua:325-327** - `receive_time` measurement (socket read)
- **nats.lua:652-677** - `waitForResponse()` loop that reads socket during send
- **nats.lua:779-804** - `processMsg()` handling `_INBOX.` responses
- **nats-grid.lua:397-404** - `networkSendTime` measurement
- **nats-grid.lua:538-557** - `networkReceiveTime` measurement
- **nats-grid.lua:273,444** - `totalSendTime` wall-clock measurement
- **nats-grid.lua:533,635** - `totalReceiveTime` wall-clock measurement
- **nats.lua:942-977** - `printStats()` aggregation

---

## STRICT RULE: No Go Legacy Code

**ABSOLUTELY NO LEGACY CODE FROM GO ALLOWED!!!!**

Lua NATS client does NOT work like Go NATS:

- No goroutines needed (Lua coroutines block and return)
- No channels needed (direct function calls return values)
- No mutexes needed (single-threaded execution)
- No async I/O wrappers needed (blocking socket reads return directly)

## Error Handling Pattern

**All public functions return a single value - response or error object:**

**Key patterns:**

1. Single return value - no `nil, err` tuples
2. Check `response.error` (boolean) to detect errors
3. Access `response.message` for error text
4. Access `response.code` for error code
5. Use `E.err(E.errStreamNotFound, E.codeStreamNotFound)` to create error objects

**No fallbacks or defaults:

## Parameter Naming Convention

**Server-Facing vs Internal Parameters**

### Server-Facing Parameters (sent to NATS server)

These parameters are converted and sent to the NATS server in API requests:

- `expires_seconds` → Converted to nanoseconds, sent as `expires` in pull request JSON
- `idle_heartbeat_seconds` → Converted to nanoseconds, sent as `idle_heartbeat` in pull request JSON

Code location: nats-pull-consumer.lua:109-115

```lua
local reqExpiresNanos = (opt.expires_seconds or 5) * 1e9
local reqIdleHeartbeatNanos = opt.idle_heartbeat_seconds and opt.idle_heartbeat_seconds * 1e9
local req = {batch = needed, max_bytes = opt.max_bytes, no_wait = opt.wait == false,
             expires = reqExpiresNanos, idle_heartbeat = reqIdleHeartbeatNanos}
```

### Internal Lua Parameters (used only locally)

These parameters are used internally in Lua code:

- `timeout` → Local timeout for `processIncoming()` loop (line 134, 146 in nats-pull-consumer.lua)
- NOT sent to server

**Key distinction:**

- `expires_seconds` → Server controls when pull request expires
- `timeout` → Local Lua code controls how long to wait for messages to arrive

**Why this matters:**

When you read code that says "if internal then -> timeout", it means:

- Internal callers use `timeout` parameter directly
- External facing APIs use `expires_seconds` which gets converted and sent to server**

## Two-Loop Pattern: processIncoming() and waitForResponse()

**CRITICAL: Understanding the dual-loop architecture**

NATS uses a two-loop pattern for message processing:

### Loop 1: processIncoming() - "Process All Pending"

Location: nats.lua:323-368

```lua
repeat
    readData, sockErr, readLen = self.sock:receive(api.CRLF, timeout)
    -- ... process message ...
    if timeout == 0 and parserState.state == 0 then
        return  -- Exit: buffer empty, non-blocking read
    end
until false
```

**Purpose:** Process ALL pending messages in the socket buffer in one call.

**Behavior:**

- With `timeout > 0`: Blocks until data arrives, then loops to process all pending messages
- With `timeout = 0`: Non-blocking - processes available messages, returns when buffer empty
- Parser calls `processMsg()` callback for each message
- `processMsg()` checks if subject starts with `_INBOX.`, extracts token, updates `responseMap[token].done = true`

### Loop 2: waitForResponse() - "Wait for Specific Response"

**Purpose:** Loop calling `processIncoming()` until the specific response arrives or timeout.

Used by: `RequestWithContext()`, `RequestMsgWithContext()`, pull consumer fetch operations.

**Why this two-loop pattern works:**

- Lua coroutines block and return - no goroutines needed
- Single-threaded - no mutexes needed
- Blocking reads return directly when data arrives - no async wrappers needed
- responseMap isolates concurrent requests by unique token
- processIncoming() efficiently drains buffer (no partial reads)
- waitForResponse() provides semantic waiting for specific responses

## Subscribe Confirmation Pattern

**subscribe() uses PING/PONG for server confirmation:**

When you call `nc:subscribe("foo", callback)`, the function MUST block until the server confirms the subscription is active. Otherwise, if you publish immediately, the message might be lost because the server hasn't processed the SUB command yet.

**How it works:**

1. `subscribe()` sends `SUB foo _EMPTY_ 5\r\nPING\r\n` in one `socketWrite()` call
2. Loops calling `processIncoming()` until `pongCount` increments
3. `processPong()` increments `pongCount` when PONG arrives
4. Server processes SUB first, then responds with PONG
5. When PONG arrives, `subscribe()` returns

**Why pongCount instead of checking lastOp:**

- `processIncoming()` processes ALL pending messages in one call
- `lastOp` is the LAST message processed, not the PONG we're waiting for
- Other messages (MSG, INFO, etc.) might arrive before PONG
- `pongCount` is a counter that increments for each PONG - reliable tracking

**Why Go's kickFlusher won't work in Lua:**

Go uses:

- Background goroutine `flusher()` that continuously flushes buffer
- `kickFlusher()` signals the goroutine to flush NOW
- Asynchronous - doesn't wait for confirmation

Lua (single-threaded, no goroutines):

- Can't use background goroutines
- Need synchronous confirmation
- PING/PONG pattern provides confirmation

**Only subscribe() needs PONG confirmation:**

- `subscribe()` - Timing matters! You publish immediately after, need confirmation subscription is active
- `unsubscribe()` - Local cleanup sufficient (we mark `closed=true` immediately, `processMsg()` checks this)
- `close()` - Hard socket close, no confirmation needed
- `publish()` - Fire and forget, no confirmation needed

## Lua NATS Client Architecture

### Callback System

**Callbacks ARE needed for:**

- **Regular subscription** - When you `subscribe("foo.bar", callback)`, you want to process messages as they arrive over time. This is the pub/sub pattern where multiple messages come in asynchronously.

**Callbacks are NOT needed for:**

- **Request/Response** - Response messages arrive on `_INBOX.token` subjects. `processMsg()` checks the subject directly, extracts the token, and updates `responseMap[token].done = true`. The `_INBOX.*` subscription has `nil` callback - all work is done in `processMsg()`.

### Message Ordering

**Other messages CAN arrive between request and response:**

1. **Heartbeat messages** - NATS server sends `NATS/1.0 100 Idle Heartbeat` periodically
2. **Flow control messages** - `NATS/1.0 100 FlowControl Request` for JetStream consumer push
3. **Messages on other subscription** - If you have other active subscription, their messages arrive
4. **INFO updates** - Server may send updated INFO during connection

**How responseMap handles this:**

- Each request gets unique token in `responseMap[token] = {done, payload, error}`
- Wildcard subscription `_INBOX.*` receives ALL responses (with `nil` callback)
- `processMsg()` checks if subject starts with `_INBOX.`, extracts token, sets `responseMap[token].done = true`
- `RequestWithContext()` loops calling `processIncoming()` until its response is ready
- Other messages are processed normally (heartbeats logged, flow control handled, subscription called)

## Installing NATS Mac OS

NATS configuration file and installation instructions can be found in /nc-devops/conf/nats.conf. Conf file needs to be moved into /private/etc/nats.conf. Restart the NATS server after the configuration file is in place.

You can install NATS with brew with following command.

```bash
brew install nats-server
brew tap nats-io/nats-tools
brew install nats-io/nats-tools/nats # for nats-cli, binary in /opt/homebrew/bin/nats
```

```conf
#		<key>ProgramArguments</key>
#		<array>
#			<string>/opt/homebrew/opt/nats-server/bin/nats-server</string>
#			<string>-c</string>
#			<string>/private/etc/nats/nats.conf</string>
#		</array>
```

Previously shown line needs to be copied into the file `homebrew.mxcl.nats-server.plist` after `<array></array>`.

File locations:

- in M1: code /opt/homebrew/Cellar/nats-server/2.10.16/homebrew.mxcl.nats-server.plist
- in x64: code /usr/local/Cellar/nats-server/2.10.16/homebrew.mxcl.nats-server.plist
- it copies file on every brew services restart to /Users/<yourusername>/Library/LaunchAgents/homebrew.mxcl.nats-server.plist

After copying, do following steps.

Create folder "/private/etc/nats":

```bash
sudo mkdir -p /private/etc/nats
```

Set ownership to your user for the folder:

```bash
sudo chown -R $(whoami) /private/etc/nats/
```

Create folder for JetStream.

```bash
mkdir -p /private/etc/nats/jetstream
```

Create nats-server.log file for logging server information:

```bash
touch /private/etc/nats/nats-server.log
```

Copy proper config file from nc-devops/conf/nats/nats.conf to /private/etc/nats/:

```bash
cp /Volumes/nc/nc-devops/conf/nats/nats.conf /private/etc/nats/nats.conf
```

After completing previous steps restart nats-server:

```bash
brew services restart nats-server
```

If everything goes finem, a following command should print no errors:

```bash
sudo nats-server -c /private/etc/nats.conf
```

You can also see if server has properly boot up with a following command:

```bash
tail -f /private/etc/nats/nats-server.log
```

## Testing that NATS server works

Following commands can be run to test NATS server.

Check that connection is healthy:

```bash
nats server check connection
```

You can try messaging by opening two terminal windows.

In Window 1 you subscribe to "test" with following command:

```bash
nats sub test
```

In Window 2 you send message to "test" with following command:

```bash
nats pub test 'Hello, NATS'
```

If everything works you will receive message 'Hello, NATS'.

## NATS JetStream object store

To use an object store, ensure that JetStream is enabled. You can verify it with the following command:

```bash
nats account info
```

If authorization is on, NATS account info needs credentials. Here is an example command:

```bash
nats account info --server "nats://username:password@localhost:4222"
```

The output should be something like this:

```conf
Connection Information:

               Client ID: 6
               Client IP: 127.0.0.1
                     RTT: 64.996µs
       Headers Supported: true
         Maximum Payload: 1.0 MiB
           Connected URL: nats://127.0.0.1:4222
       Connected Address: 127.0.0.1:4222
     Connected Server ID: ND2XVDA4Q363JOIFKJTPZW3ZKZCANH7NJI4EJMFSSPTRXDBFG4M4C34K

JetStream Account Information:

           Memory: 0 B of Unlimited
          Storage: 0 B of Unlimited
          Streams: 0 of Unlimited
        Consumers: 0 of Unlimited
```

Use the following commands to use the JetStream object-store.

Delete a bucket called "work-center-test":

```bash
nats --trace object del work-center-test
```

Create a bucket called "work-center-test":

```bash
nats --trace object add work-center-test
```

Create a file called "testfile.json" and add simple contents:

```bash
echo '{ "message": "Hello world!" }' > testfile.json
```

Put a file called "testfile" to a bucket "work-center-test":

```bash
nats --trace object put work-center-test testfile.json
```

Put a file called "testfile" to the bucket and give it the name "myname":

```bash
nats --trace object put --name myname work-center-test testfile.json
```

List objects in "work-center-test":

```bash
nats --trace object ls work-center-test
```

Get file "testfile" from "work-center-test":

```bash
nats --trace object get work-center-test testfile.json
```

Get file "testfile" from "work-center-test" to a specific output path:

```bash
nats --trace object get work-center-test --output /temp/testfile.json testfile.json
```

Remove an object from the bucket "work-center-test":

```bash
nats --trace object rm work-center-test testfile.json
```

Get information about bucket "work-center-test":

```bash
nats --trace object info work-center-test
```

Watch changes for "work-center-test":

```bash
nats --trace object watch work-center-test
```

Seal "work-center-test":

```bash
nats --trace object seal work-center-test
```

Delete "work-center-test":

```bash
nats --trace object rm work-center-test
```

## Using NATS with JavaScript and nats.ws library

NATS supports two types of wildcards for matching subobjects.

### Wildcard *

Wildcard '*' matches only one level of subobject for example "object.subobject1"

### Wildcard >

'>' matches multiple levels of subjects for example "object.subobject1.subobject2"

### Subscribing to multiple subjects

Messages sent to both "testnats" and "testnats.*" can be received by creating multiple subscriptions.

## Using NATS JetStream Object Storage with JavaScript and nats.ws

Documentation for using ObjectStorage with JavaScript: <https://natsbyexample.com/examples/os/intro/deno>

To use JetStream with nats.ws library, data must be in uint8Array format. Objects can be stored, retrieved, and managed through the Object Storage API.

## Using Wireshark to monitor NATS traffic

- bi --cask wireshark (brew install)
- open Wireshark
  - select Loopback: lo0
  - add filter 'port 4222' to see only nats traffic

![alt text](image/wireshark-filter.png)

- start from nats.md: nats --trace object del work-center-test
- see the traffic in Wireshark
  - select some row
  - right click and select follow TCP stream
  - Wireshark opens a new 'follow TCP stream' -window with the traffic

![alt text](image/wireshark-follow-tcp.png)

## Pull vs Push Consumers

NATS JetStream supports two consumer types for receiving messages:

### Pull Consumers (Recommended)

Pull consumers give you **on-demand control** over message retrieval.

**Key characteristics:**

- **You request messages** - consumer waits for your pull request
- **On-demand fetching** - fetch only when you need data
- **Better flow control** - you control the rate of consumption
- **Works with any batch size** - request 1, 10, or 1000 messages
- **No push overhead** - no continuous message stream

**Fetch behavior:**

- `fetch(N)` returns when N messages arrive OR timeout expires
- Returns immediately with available messages if fewer than N exist
- Use `count` parameter to limit fetch size: `fetch(chunk_size, {count = actual_count})`

**When to use pull consumers:**

- Reading data on-demand (user requests, polling scenarios)
- Batch processing where you control the pace
- Grid/fetch operations where you know the expected count
- Most common use cases

### Push Consumers

Push consumers continuously deliver messages to a subscription.

**Key characteristics:**

- **Server pushes messages** - continuous delivery to your client
- **Callback-based** - messages arrive as soon as they're published
- **Requires `deliver_subject`** - inbox for server to push messages to
- **Automatic flow control** - server manages delivery rate

**When to use push consumers:**

- Real-time updates (chat, presence, live feeds)
- Event-driven architectures
- Low-latency requirements
- When you need immediate notification of new messages

### Critical Rule: Request Correct Count

**ALWAYS request the number of messages that actually exist!**

**Why this matters:**

- `fetch(N)` waits up to timeout for N messages
- If fewer than N exist, you wait the full timeout period
- Requesting the actual count returns immediately (~1-2ms)

### Common Mistakes

1. **Requesting more than exists:** Always request exactly the number of messages that exist, or use the count parameter.

2. **Not counting chunks vs rows:** Grid chunks use message count, not row count (e.g., 250 rows = 2 chunks).

3. **Using heartbeat without matching expiry:** Heartbeat must be <= half of expiry time.

## NATS Server Configuration

### Critical Timeout Settings

Location: `/Volumes/nc/nc-devops/conf/nats/nats.conf` (copied to `/private/etc/nats/nats.conf`)

```hocon
# Connection timeouts - prevent slow consumer disconnect on large message batches
write_deadline: "120s"    # Server-side timeout for writing to client (default: 10s)
ping_interval: "2m"       # Client ping interval (default: 2m)
ping_max: 2               # Max unanswered pings before disconnect (default: 2)

jetstream {
    store_dir: "/private/etc/nats"
    max_mem: 1G
    max_file: 10G
    limits {
        max_request_batch: 1000   # Max fetch batch size (default: unlimited)
        max_ack_pending: 10000    # Max pending acks (default: unlimited)
    }
}
```

### Why These Settings Matter

| Setting | Default | Problem | Fix |
|---------|---------|---------|-----|
| `write_deadline` | 10s | Server aborts sending large batches (>100MB) | 120s for 400MB+ batches |
| `max_pending` | 64MB | Server connection closed when buffered data exceeds limit | 512MB for large batches |
| `max_request_batch` | unlimited | Client may fetch too many messages | 1000 for safe limit |
| `max_ack_pending` | unlimited | Max pending acks in explicit ack mode | 10000 for high throughput |

### Timeout Hierarchy

```text
Client-Side Timeouts:
├── default_timeout = 5s             (nats.lua:250, socket read timeout)
├── connect_timeout = 5s             (nats.lua:250, connection establishment)
├── expiresDefaultTimeout = 55s      (nats-ordered-consumer.lua:23)
│   └── Must be < TCP socket timeout (60s) - OS system error code 60
├── TCP socket timeout = 60s         (OS level, system error code)
│   └── OS network timeout (macOS BSD sockets: ETIMEDOUT error 60)
└── fetch expires_seconds = 120s     (nats-grid.lua:369, for large batches > 100)
    └── Requires server write_deadline >= 120s

Server-Side Timeouts (nats.conf):
├── write_deadline = 10s (default)  → Too short for 400MB batches!
├── ping_interval = 2m (default)
├── ping_max = 2 (default)
└── JetStream limits:
    ├── max_request_batch = unlimited (default)
    └── max_ack_pending = unlimited (default)

FIXED CONFIGURATION (/Volumes/nc/nc-devops/conf/nats/nats.conf):
├── write_deadline = 120s  → Gives client time to receive large batches
├── max_pending = 512MB  → Connection-level buffer limit (default: 64MB)
├── max_request_batch = 1000  → Safe upper limit for fetch size
└── max_ack_pending = 10000  → Allows more in-flight acks
```

### max_pending - Connection Buffer Limit

**Purpose:** Maximum number of bytes buffered for a connection (server-side limit)

**Default:** `64MB` (67108864 bytes)

**Error when exceeded:** `Slow Consumer Detected: MaxPending of 67108864 Exceeded`

**Configuration:**

```hocon
max_pending: 5368709120  # 512MB in bytes (512 * 1024 * 1024)
```

**Note:** `max_pending` uses raw bytes, NOT a quoted string with units

**Code location:** Server configuration, applies to all client connections

**Hot reloadable:** Yes

### JetStream Consumer max_bytes

**Purpose:** Maximum bytes the consumer can have in-flight (pending acknowledgment)

**Configuration (consumer creation):**

```lua
local consumerConfig = {
    ack_policy = "none",
    mem_storage = true,
    max_bytes = 104857600,  -- 100MB in-flight limit
    -- or max_bytes = -1 for unlimited
}
```

**Related to:** JetStream pull consumer pending message limit, NOT the same as server `max_pending`

### Client-Side PendingLimits (Regular Subscriptions)

**Purpose:** Set limits on incoming messages for a subscription (NOT JetStream)

**Go example:**

```go
sub.SetPendingLimits(1000, 5*1024*1024)  -- 1000 messages or 5MB
```

**Lua client:** Check if your NATS client library supports `setPendingLimits` for regular subscriptions

**Note:** This is different from JetStream consumer configuration

### Client Code Values

**`nats-ordered-consumer.lua:23`**

```lua
local expiresDefaultTimeout = 55 -- Must be less than TCP socket timeout (60 sec)
```

**`nats-grid.lua:368-370`**

```lua
if option.message_count and option.message_count > 100 then
    fetchOpt.expires_seconds = 120  -- Server must allow this via write_deadline
end
```

**`nats.lua:250` - Default Connection Options**

```lua
local obj = {url = nats.default_url, servers = {}, name = nil, verbose = false,
    pedantic = false, user = nil, password = nil, token = nil,
    default_timeout = 5,      -- Socket timeout for read operations (seconds)
    connect_timeout = 5,      -- Connection establishment timeout (seconds)
    headers = false, max_payload = 1048576}
```

**`nats-grid-test.lua:73` - Test Connection Options**

```lua
local connectOption = {
    host = "localhost",
    port = 4222,
    connect_timeout = 3,      -- Test uses shorter connection timeout
    default_timeout = 0.5,    -- Test uses shorter socket timeout
    trace = trace,
    print_error = true
}
```

### Applying Configuration Changes

```bash
# Edit source config
sudo micro /Volumes/nc/nc-devops/conf/nats/nats.conf

# Test config file syntax
nats-server -c /Volumes/nc/nc-devops/conf/nats/nats.conf -T

# Copy to active location
cp /Volumes/nc/nc-devops/conf/nats/nats.conf /private/etc/nats/nats.conf

# Restart service
brew services restart nats-server

# Verify server is running
ps aux | grep nats-server | grep -v grep

# Check log
tail -20 /private/etc/nats/nats-server.log
```

### Verifying Configuration on Running Server

To verify that configuration changes are active on the running server:

```bash
# Check max_pending and write_deadline values
curl -s http://localhost:8222/varz | python3 -c "import sys, json; d=json.load(sys.stdin); print('max_pending:', d.get('max_pending', 'NOT FOUND')); print('write_deadline:', d.get('write_deadline', 'NOT FOUND'))"

# Expected output (in nanoseconds):
# max_pending: 53687091200  (512MB)
# write_deadline: 120000000000  (120s)

# Alternative: View full server info
curl -s http://localhost:8222/varz | python3 -m json.tool | less
```

### Common Issues

**Issue: `Slow Consumer Detected: MaxPending of 67108864 Exceeded`**

- Cause: Server's `max_pending` limit (default 64MB) exceeded when buffering data for connection
- Fix: Set `max_pending: "512MB"` in server config
- Alternative: Reduce batch size or process messages faster

**Issue: Connection reset during large fetch (789+ messages, 400MB+)**

- Cause: `write_deadline` (default 10s) expires before client receives all data
- Fix: Set `write_deadline: "120s"` in server config

**Issue: ETIMEDOUT: '60: Operation timed out'**

- Cause: Client `expires_seconds` > TCP socket timeout
- Fix: Ensure `expiresDefaultTimeout = 55s` stays < 60s

**Issue: "slow consumer" warnings**

- Cause: Client not consuming fast enough for `write_deadline`
- Fix: Increase `write_deadline` or reduce batch size

## Pings, Heartbeats, and ACKs

### Why These Don't Fix Slow Consumer Issues

For large batch receives (789+ messages, 400MB+), the problem is **client read speed vs server write deadline**. Pings, heartbeats, and ACKs don't help because they address different problems.

### Pings (Connection Keep-Alive)

**Purpose:** Detect dead connections, NOT message delivery speed.

**Server config:**

```hocon
ping_interval: "2m"    # Server sends PING every 2 minutes
ping_max: 2            # Disconnect after 2 missed PINGs
```

**How it works:**

- Server sends PING, expects PONG within `ping_interval * ping_max` time
- If client doesn't respond, server assumes connection is dead
- Only verifies connection is ALIVE, not how fast client reads data

**Code location:** `nats.lua:266` - `ping_count`, `pong_count` tracked in stats

### Heartbeats (Idle Connection Detection)

**Purpose:** Detect STALLED message delivery when no messages arriving.

**Grid config** (`nats-grid-config.lua:39`):

```lua
gridConfig.consumer_idle_heartbeat = 5 * 1e9  -- 5 seconds in nanoseconds (server format)
```

**Ordered consumer default** (`nats-ordered-consumer.lua:282-283`):

```lua
if not fetchOpt.idle_heartbeat_seconds then
    fetchOpt.idle_heartbeat_seconds = 5  -- Client-side format (seconds)
end
```

**How it works:**

- Server sends `100 Idle Heartbeat` status when NO messages being delivered
- Client breaks if heartbeat timeout exceeded (`2x idle_heartbeat`)
- **Only fires when IDLE** - NOT sent during active message delivery

**Code locations:**

- `nats-pull-consumer.lua:84-87, 148-149, 191-193` - Heartbeat tracking
- `nats-pull-consumer.lua:119` - Convert to nanoseconds for server
- `nats.lua:831` - `heartbeat_count` tracked in stats

**Validation rule** (`nats-pull-consumer.lua:115-116`):

```lua
if opt.idle_heartbeat_seconds and opt.idle_heartbeat_seconds * 2 > (opt.expires_seconds or 5) then
    return self:err(E.errCodeInvalidArgument, "idle_heartbeat_seconds must be <= half of expires_seconds")
end
```

### ACKs (Message Acknowledgment)

**Purpose:** Tell server message was processed, affects delivery tracking.

**Grid config** (`nats-grid-config.lua:35`):

```lua
gridConfig.consumer_ack_policy = "none"
```

**ACK policies:**

| Policy | Behavior | Impact on Large Receive |
|--------|----------|------------------------|
| `none` | No ACKs needed | **Fastest** - server fires all messages |
| `explicit` | Client ACKs each msg | **Slower** - adds round-trip delay |
| `all` | Auto-ACK on receive | **Slower** - adds processing overhead |

**Code location:** `nats-ordered-consumer.lua:66` - Applied to consumer config

**Why ACKs won't help (and would make it worse):**

```text
Current (ack_policy = "none"):
Server → [msg1][msg2][msg3]...[msg789] → Client (fast, continuous)

With explicit ACKs:
Server → [msg1] → Client → ACK → Server → [msg2] → Client → ACK ...
(round-trip delay for EACH message, MUCH slower)
```

ACKs add MORE work for the client, making it SLOWER to read messages, causing `write_deadline` to expire FASTER.

### What Actually Fixes Large Receive Issues

| Problem | Mechanism | Helps? |
|---------|-----------|--------|
| Connection dead | Pings | Yes ✓ |
| Stalled delivery | Heartbeats | Yes ✓ (when idle) |
| Slow consumer | ACKs | No ✗ (makes it worse) |
| **Slow consumer** | **`write_deadline`** | **Yes ✓** |

The `write_deadline: "120s"` server setting is the ONLY fix for large batch receives. It gives the client enough time to receive 400MB+ of data before the server aborts as "slow consumer".

### Summary: Timeout Hierarchy

```text
┌─────────────────────────────────────────────────────────────────────┐
│                    CONNECTION HEALTH                                │
├─────────────────────────────────────────────────────────────────────┤
│ Pings (ping_interval, ping_max)     → Detect dead connections      │
│ Heartbeats (idle_heartbeat)         → Detect stalled delivery      │
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                    MESSAGE DELIVERY SPEED                           │
├─────────────────────────────────────────────────────────────────────┤
│ write_deadline (server)              → Abort slow consumers        │
│ expires_seconds (client)             → Fetch timeout               │
│ default_timeout (client socket)      → Socket read timeout         │
│ TCP socket timeout (OS)              → ETIMEDOUT error 60          │
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                    MESSAGE TRACKING                                 │
├─────────────────────────────────────────────────────────────────────┤
│ ACKs (ack_policy)                    → Message processed status    │
│ max_ack_pending (server)             → Max unacknowledged msgs      │
└─────────────────────────────────────────────────────────────────────┘
```

## Pending Message Refresh Strategy

### The Problem

When using ordered consumers, the cached `num_pending` value can become **stale**:

1. **Long delay between fetches** - Consumer info was cached minutes ago
2. **Messages consumed elsewhere** - Another consumer read pending messages
3. **Race condition** - New messages arrive after cache but before fetch

### The Solution: Two-Phase Fetch with Refresh

The ordered consumer implements a **smart pending refresh** strategy.

### How It Works

**Phase 1: Quick Fetch (0.05s timeout)**

- Attempts to fetch messages rapidly
- If messages arrive quickly, returns them immediately
- If timeout occurs, proceeds to Phase 2

**Phase 2: Refresh & Verify**

- Fetches fresh `consumerInfo` from NATS server
- If `num_pending < 1`: Returns timeout error (no messages available)
- If `num_pending > 0`: Returns what was received (caller can retry)
- If got messages but `#msgs < batch`: Refresh again to check for more

### When This Helps

**1. Large Batch Operations** - Request more messages than available, refresh shows 0 pending, returns immediately without timeout.

**2. Long Delays Between Calls** - Fresh info gives accurate count after delays, returns immediately if empty.

**3. Race Conditions** - If another consumer reads messages while fetching, refresh pending to verify and don't set exhausted if more messages available.

### Performance Impact

**Before:** Stale cache → 5 second timeout. Total time: **5+ seconds**

**After:** Quick fetch (0.05s) + Refresh (~0.01s). Total time: **~0.06 seconds** (~80x faster for empty/stale scenarios)

### Key Takeaways

1. **Always use fresh info for pending counts** - Cached values become stale
2. **Short timeout first** - Get messages quickly if available
3. **Refresh after fetch** - Verify actual pending count
4. **Handle partial results** - Don't set exhausted if `#msgs < batch` and pending > 0
