fix: bound upstream request failures (#10)

Classify upstream failures before retrying, enforce one end-to-end
request budget, and preserve Modbus exception responses. Reads keep one
safe transport retry while ambiguous writes fail without being repeated.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 2fa557c2-9b1f-4cf7-8e68-3a5141cd5261

Author: tma

README.md

@@ -33,12 +33,28 @@ All configuration is via environment variables:
 | `MODBUS_CACHE_TTL` | Cache time-to-live | `10s` |
 | `MODBUS_CACHE_SERVE_STALE` | Serve stale data on upstream error | `false` |
 | `MODBUS_READONLY` | Read-only mode: `false`, `true`, `deny` | `true` |
-| `MODBUS_TIMEOUT` | Upstream connection timeout | `10s` |
-| `MODBUS_REQUEST_DELAY` | Delay after each upstream request | `0` (disabled) |
+| `MODBUS_ATTEMPT_TIMEOUT` | Per-attempt upstream socket timeout | `10s` |
+| `MODBUS_TIMEOUT` | Deprecated alias for `MODBUS_ATTEMPT_TIMEOUT` | unset |
+| `MODBUS_REQUEST_TIMEOUT` | Total request budget, including coalescing, queueing, reconnect, retry, and pacing | `30s` |
+| `MODBUS_REQUEST_DELAY` | Minimum interval between successful upstream requests | `0` (disabled) |
 | `MODBUS_CONNECT_DELAY` | Silent period after connecting to upstream | `0` (disabled) |
 | `MODBUS_SHUTDOWN_TIMEOUT` | Graceful shutdown timeout | `30s` |
 | `LOG_LEVEL` | Log level: `INFO`, `DEBUG` | `INFO` |
 
+The end-to-end request budget always caps each individual attempt. A full read
+retry budget needs room for two attempt timeouts, two connect delays, request
+pacing, and any dial time. Pacing is a context-aware pre-wire wait charged to
+the next request's budget; it never delays an already received response.
+Existing configurations may keep using `MODBUS_TIMEOUT` during migration.
+When both attempt timeout variables are set, their parsed durations must match
+or startup fails. Neither setting changes `MODBUS_REQUEST_TIMEOUT`.
+
+Downstream exceptions preserve genuine upstream Modbus exception responses with
+nonzero exception codes. Upstream transport, framing or malformed exception
+failures and total request deadlines map to gateway target failed to respond
+(`0x0B`). Local internal failures map to server failure (`0x04`), while local
+validation uses the standard validation exception codes.
+
 `/mbproxy -health` performs an internal upstream connectivity check and does not open a separate local TCP health port.
 
 ### Read-Only Modes
@@ -77,7 +93,8 @@ services:
       MODBUS_CACHE_TTL: "10s"
       MODBUS_CACHE_SERVE_STALE: "false"
       MODBUS_READONLY: "true"
-      MODBUS_TIMEOUT: "10s"
+      MODBUS_ATTEMPT_TIMEOUT: "10s"
+      MODBUS_REQUEST_TIMEOUT: "30s"
       MODBUS_REQUEST_DELAY: "0"
       MODBUS_CONNECT_DELAY: "0"
       MODBUS_SHUTDOWN_TIMEOUT: "30s"
@@ -137,9 +154,9 @@ docker run --rm -v $(pwd):/app -w /app golang:1.24 go test ./...
 - **Key format**: values are cached per register/coil as `{slave_id}:{function_code}:{address}`
 - **Read requests**: Served from cache only if every register/coil in the requested range is present and not expired
 - **Cache misses**: If any value in the requested range is missing or expired, the full range is fetched from upstream and decomposed into per-register/coil cache entries
-- **Write requests**: Forwarded to upstream (if allowed), then invalidate the written address range so overlapping cached reads cannot return stale values
-- **Request coalescing**: Multiple identical range requests during a cache miss share a single upstream fetch using `{slave_id}:{function_code}:{start_address}:{quantity}` as the coalescing key
-- **Stale fallback**: If enabled, expired entries are retained and can be served when upstream requests fail
+- **Write requests**: Before an allowed write is forwarded, its generation is incremented and the written range is invalidated. The generation is incremented and the range invalidated again after every outcome, so neither older reads nor reads that execute in the write scheduling window can leave pre-write values cached.
+- **Request coalescing**: Multiple identical range requests in the same write generation share a single upstream fetch using `{write_generation}:{slave_id}:{function_code}:{start_address}:{quantity}` as the coalescing key
+- **Stale fallback**: If enabled, expired entries are retained and can be served when upstream transport requests fail. Upstream Modbus exceptions are never replaced with stale data.
 
 ## License
 

SPEC.md

@@ -41,8 +41,9 @@ Many Modbus devices (inverters, meters, battery systems) have limited polling ca
 - Connect to downstream Modbus device via TCP/IP only
 - Support multiple slave IDs through single connection
 - Support clients requesting different slave IDs through the proxy
-- Auto-reconnect on connection failure (unlimited retries, no backoff)
-- Request pacing: configurable delay between upstream requests to prevent overwhelming slow devices
+- Reconnect after transport failures; retry reads once but never retry ambiguous writes
+- Preserve upstream Modbus exceptions without reconnecting
+- Request pacing: configurable minimum interval between successful upstream requests
 - TCP keep-alive enabled (30s interval) for connection health monitoring
 - Connect delay: optional silent period after establishing connection for device settling
 
@@ -57,7 +58,7 @@ Values are cached per register/coil:
 
 Request coalescing still uses the requested range as its key:
 ```
-{slave_id}:{function_code}:{start_address}:{quantity}
+{write_generation}:{slave_id}:{function_code}:{start_address}:{quantity}
 ```
 
 #### Cache Entry
@@ -71,23 +72,24 @@ type CacheEntry struct {
 
 #### Cache Behavior
 - **Read Operations**: Check the per-register/coil cache first. Return from cache only if every value in the requested range is present and not expired.
-- **Cache Misses**: If any value in the requested range is missing or expired, fetch the full requested range from upstream, then decompose the response into per-register/coil cache entries.
-- **Write Operations**: Always forward to the device when writes are allowed, then invalidate each cached register/coil in the written address range. This prevents overlapping cached read ranges from serving stale values after frequent writes.
+- **Cache Misses**: If any value in the requested range is missing or expired, fetch the full requested range from upstream, then decompose the response into per-register/coil cache entries only if the write generation is unchanged.
+- **Write Operations**: Before forwarding an allowed write, increment the write generation and invalidate each cached register/coil in the written address range. After every write outcome, increment and invalidate again so a read that entered the new generation but executed before the write cannot leave a pre-write value cached. This preserves ambiguous-write invalidation without holding the cache state lock across upstream I/O.
 - **TTL**: Configurable (default: 10 seconds)
 - **Cleanup**: Time-based expiration. Expired entries are removed during cleanup unless stale serving is enabled.
 - **Staleness**: Option to serve stale data on upstream failure (default: off). When enabled, expired entries are retained so they remain available for fallback.
 
 ### Request Coalescing
-- Identical in-flight range requests are coalesced (same slave_id, function, address, quantity)
+- Identical in-flight range requests are coalesced within the same write generation
 - Second request arriving while first is pending will wait for and share the first's response
 - Prevents thundering herd on cache miss
 
 ### Request Pacing
-- Configurable delay after each successful upstream request
+- Configurable minimum interval measured from each successful upstream response
 - Protects slow Modbus devices that cannot handle rapid-fire requests
-- Delay is context-aware: cancelled if the request context is cancelled
-- Only applied after successful requests (not during error recovery/reconnection)
-- Logged at DEBUG level when applied
+- Enforced as a context-aware pre-wire wait for the next request
+- Consumes the next request's end-to-end budget and never delays or reclassifies the completed request
+- Not reapplied between a failed read attempt and its retry
+- Logged at DEBUG level when a request waits for its slot
 
 ### 4. Read-Only Mode
 Three modes:
@@ -110,12 +112,28 @@ Three modes:
 | `MODBUS_CACHE_TTL` | Cache time-to-live | `10s` | `10s`, `1m`, `500ms` |
 | `MODBUS_CACHE_SERVE_STALE` | Serve stale data on upstream error | `false` | `true`, `false` |
 | `MODBUS_READONLY` | Read-only mode | `true` | `false`, `true`, `deny` |
-| `MODBUS_TIMEOUT` | Upstream connection timeout | `10s` | `5s`, `30s` |
-| `MODBUS_REQUEST_DELAY` | Delay after each upstream request | `0` (disabled) | `100ms`, `500ms` |
+| `MODBUS_ATTEMPT_TIMEOUT` | Per-attempt upstream socket timeout | `10s` | `10s`, `30s` |
+| `MODBUS_TIMEOUT` | Deprecated alias for `MODBUS_ATTEMPT_TIMEOUT` | unset | `10s`, `30s` |
+| `MODBUS_REQUEST_TIMEOUT` | End-to-end request budget | `30s` | `30s`, `1m` |
+| `MODBUS_REQUEST_DELAY` | Minimum interval between successful upstream requests | `0` (disabled) | `100ms`, `500ms` |
 | `MODBUS_CONNECT_DELAY` | Silent period after connecting to upstream | `0` (disabled) | `500ms`, `2s` |
 | `MODBUS_SHUTDOWN_TIMEOUT` | Graceful shutdown timeout | `30s` | `10s`, `60s` |
 | `LOG_LEVEL` | Log level | `INFO` | `INFO`, `DEBUG` |
 
+`MODBUS_ATTEMPT_TIMEOUT` is preferred. `MODBUS_TIMEOUT` remains accepted as a
+deprecated migration alias. If both are set, their parsed durations must be
+equal or configuration loading fails. These variables do not set or override
+`MODBUS_REQUEST_TIMEOUT`.
+
+The end-to-end budget caps every individual attempt. Retaining the read retry
+requires enough budget for two attempt timeouts, two connect delays, request
+pacing, and dial time. Pacing consumes the next request's budget before its wire
+attempt. Genuine upstream Modbus exception responses keep their nonzero
+exception code downstream. Upstream transport or framing failures, malformed
+exception responses, and total request deadlines map to `0x0B`; local internal
+failures map to `0x04`; local validation keeps the standard validation exception
+codes.
+
 The container health check runs `mbproxy -health`, which performs an internal upstream connectivity check without binding a separate local TCP port.
 
 ## Implementation Details
@@ -222,12 +240,12 @@ The cache also exposes `Coalesce(ctx, rangeKey, fetch)` for request coalescing.
 3. **For reads**:
    - Check every per-register/coil cache key in the requested range
    - If all values are present and valid, reassemble and return the Modbus response
-   - On any miss or expired value: coalesce identical in-flight range requests, then forward to upstream device
-   - Decompose successful upstream responses into per-register/coil cache entries
+   - On any miss or expired value: coalesce identical in-flight range requests within the current write generation, then forward to upstream
+   - Decompose successful upstream responses into per-register/coil cache entries only if the generation is unchanged
    - Return response to client
 4. **For writes**:
    - Check readonly mode
-   - If allowed: forward to upstream, then invalidate every cached register/coil in the written address range
+   - If allowed: increment the write generation and invalidate every cached register/coil in the written address range before forwarding upstream
    - Return response
 
 ## Logging
@@ -242,7 +260,7 @@ level=INFO msg="starting proxy" listen=:5502 upstream=192.168.1.100:502
 level=DEBUG msg="cache hit" slave_id=1 func=0x03 addr=0 qty=10
 level=DEBUG msg="cache miss" slave_id=1 func=0x03 addr=0 qty=10
 level=DEBUG msg="upstream request completed" slave_id=1 func=0x03 addr=0 qty=10 duration=15ms
-level=DEBUG msg="applying request delay" delay=100ms
+level=DEBUG msg="waiting for upstream request slot" delay=100ms
 level=DEBUG msg="applying connect delay" delay=500ms
 level=WARN msg="upstream error, serving stale" slave_id=1 error="timeout"
 level=INFO msg="shutting down"

internal/cache/cache.go

@@ -3,6 +3,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -36,9 +37,10 @@ type Cache struct {
 }
 
 type inflightRequest struct {
-	done   chan struct{}
-	result []byte
-	err    error
+	done      chan struct{}
+	result    []byte
+	err       error
+	followers int
 }
 
 // New creates a new cache with the specified default TTL.
@@ -211,51 +213,65 @@ func (c *Cache) DeleteRange(slaveID byte, functionCode byte, startAddr uint16, q
 // Other callers with the same key wait for and share the first caller's result.
 // This handles request coalescing only — it does not interact with cache storage.
 func (c *Cache) Coalesce(ctx context.Context, key string, fetch func(context.Context) ([]byte, error)) ([]byte, error) {
-	c.inflightMu.Lock()
-	if req, ok := c.inflight[key]; ok {
+	for {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+
+		c.inflightMu.Lock()
+		req, ok := c.inflight[key]
+		if !ok {
+			req = &inflightRequest{done: make(chan struct{})}
+			c.inflight[key] = req
+		} else {
+			req.followers++
+		}
 		c.inflightMu.Unlock()
-		// Wait for the in-flight request to complete
-		select {
-		case <-req.done:
-			if req.err != nil {
-				return nil, req.err
+
+		if ok {
+			if err := ctx.Err(); err != nil {
+				return nil, err
+			}
+			select {
+			case <-req.done:
+				if err := ctx.Err(); err != nil {
+					return nil, err
+				}
+				if req.err != nil {
+					if errors.Is(req.err, context.Canceled) || errors.Is(req.err, context.DeadlineExceeded) {
+						continue
+					}
+					return nil, req.err
+				}
+				data := make([]byte, len(req.result))
+				copy(data, req.result)
+				return data, nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
 			}
-			// Return a copy
-			data := make([]byte, len(req.result))
-			copy(data, req.result)
-			return data, nil
-		case <-ctx.Done():
-			return nil, ctx.Err()
 		}
-	}
 
-	// Create new in-flight request
-	req := &inflightRequest{
-		done: make(chan struct{}),
-	}
-	c.inflight[key] = req
-	c.inflightMu.Unlock()
+		data, err := fetch(ctx)
+		if err == nil {
+			err = ctx.Err()
+		}
 
-	// Fetch the data
-	data, err := fetch(ctx)
+		req.result = data
+		req.err = err
 
-	// Store result for waiters
-	req.result = data
-	req.err = err
+		c.inflightMu.Lock()
+		delete(c.inflight, key)
+		c.inflightMu.Unlock()
+		close(req.done)
 
-	// Clean up and notify waiters
-	c.inflightMu.Lock()
-	delete(c.inflight, key)
-	c.inflightMu.Unlock()
-	close(req.done)
+		if err != nil {
+			return nil, err
+		}
 
-	if err != nil {
-		return nil, err
+		result := make([]byte, len(data))
+		copy(result, data)
+		return result, nil
 	}
-
-	result := make([]byte, len(data))
-	copy(result, data)
-	return result, nil
 }
 
 // cleanupOnce runs a single cleanup pass, removing expired entries.

internal/cache/cache_test.go

@@ -2,6 +2,8 @@ package cache
 
 import (
 	"context"
+	"errors"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -310,10 +312,10 @@ func TestCache_ContextCancellation(t *testing.T) {
 
 	// Start a slow fetch
 	go func() {
-		c.Coalesce(ctx, "key1", func(ctx context.Context) ([]byte, error) {
+		_, _ = c.Coalesce(ctx, "key1", func(ctx context.Context) ([]byte, error) {
 			close(fetchStarted)
-			time.Sleep(time.Second)
-			return []byte("fetched"), nil
+			<-ctx.Done()
+			return nil, ctx.Err()
 		})
 	}()
 
@@ -334,6 +336,88 @@ func TestCache_ContextCancellation(t *testing.T) {
 	cancel()
 }
 
+func TestCache_CanceledFollowerDoesNotReturnReadyResult(t *testing.T) {
+	c := New(time.Second, false)
+	defer c.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	req := &inflightRequest{
+		done:   make(chan struct{}),
+		result: []byte("stale success"),
+	}
+	close(req.done)
+	c.inflight["key1"] = req
+
+	data, err := c.Coalesce(ctx, "key1", func(context.Context) ([]byte, error) {
+		return []byte("should not run"), nil
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected canceled follower, got data=%q err=%v", data, err)
+	}
+}
+
+func TestCache_LiveFollowerRetriesAfterLeaderDeadline(t *testing.T) {
+	c := New(time.Second, false)
+	defer c.Close()
+
+	leaderCtx, cancelLeader := context.WithCancel(context.Background())
+	leaderStarted := make(chan struct{})
+	leaderDone := make(chan error, 1)
+	go func() {
+		_, err := c.Coalesce(leaderCtx, "key1", func(ctx context.Context) ([]byte, error) {
+			close(leaderStarted)
+			<-ctx.Done()
+			return []byte("expired success"), nil
+		})
+		leaderDone <- err
+	}()
+	<-leaderStarted
+
+	followerDone := make(chan struct{})
+	var followerData []byte
+	var followerErr error
+	var followerFetches atomic.Int32
+	go func() {
+		followerData, followerErr = c.Coalesce(context.Background(), "key1", func(context.Context) ([]byte, error) {
+			followerFetches.Add(1)
+			return []byte("fresh"), nil
+		})
+		close(followerDone)
+	}()
+
+	deadline := time.Now().Add(time.Second)
+	for {
+		c.inflightMu.Lock()
+		req := c.inflight["key1"]
+		joined := req != nil && req.followers == 1
+		c.inflightMu.Unlock()
+		if joined {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("second caller did not join the live leader")
+		}
+		runtime.Gosched()
+	}
+
+	cancelLeader()
+	if err := <-leaderDone; !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected canceled leader, got %v", err)
+	}
+	select {
+	case <-followerDone:
+	case <-time.After(time.Second):
+		t.Fatal("live follower did not retry after leader cancellation")
+	}
+	if followerErr != nil || string(followerData) != "fresh" {
+		t.Fatalf("unexpected follower result data=%q err=%v", followerData, followerErr)
+	}
+	if followerFetches.Load() != 1 {
+		t.Fatalf("follower fetch ran %d times", followerFetches.Load())
+	}
+}
+
 func TestCache_DataIsolation(t *testing.T) {
 	c := New(time.Second, false)
 	defer c.Close()