test: cover proxy cache miss and stale fallback

Update README and SPEC for the per-register cache design, then
add proxy-level tests for miss -> fetch -> store -> hit and stale
fallback on upstream errors. Introduce a small upstream client
interface so tests can use a mock upstream without a real Modbus
connection.

Author: tma

README.md

@@ -132,10 +132,12 @@ docker run --rm -v $(pwd):/app -w /app golang:1.24 go test ./...
 
 ## Cache Behavior
 
-- **Key format**: `{slave_id}:{function_code}:{start_address}:{quantity}`
-- **Read requests**: Served from cache if available and not expired
-- **Write requests**: Forwarded to upstream (if allowed), exact matching cache entries invalidated
-- **Request coalescing**: Multiple identical requests during a cache miss share a single upstream fetch
+- **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
 
 ## License
 

SPEC.md

@@ -49,28 +49,36 @@ Many Modbus devices (inverters, meters, battery systems) have limited polling ca
 ### 3. In-Memory Cache
 
 #### Cache Key Structure
+
+Values are cached per register/coil:
+```
+{slave_id}:{function_code}:{address}
+```
+
+Request coalescing still uses the requested range as its key:
 ```
 {slave_id}:{function_code}:{start_address}:{quantity}
 ```
 
 #### Cache Entry
 ```go
 type CacheEntry struct {
-    Data      []byte
+    Data      []byte        // one register (2 bytes) or one coil/input bit (1 byte: 0 or 1)
     Timestamp time.Time
     TTL       time.Duration
 }
 ```
 
 #### Cache Behavior
-- **Read Operations**: Check cache first, return if valid (not expired)
-- **Write Operations**: Always forward to device, invalidate exact matching cache entries (same slave_id, function_code, start_address, quantity)
+- **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.
 - **TTL**: Configurable (default: 10 seconds)
-- **Cleanup**: Time-based expiration (entries removed when TTL expires)
-- **Staleness**: Option to serve stale data on upstream failure (default: off)
+- **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 requests are coalesced (same slave_id, function, address, quantity)
+- Identical in-flight range requests are coalesced (same slave_id, function, address, quantity)
 - Second request arriving while first is pending will wait for and share the first's response
 - Prevents thundering herd on cache miss
 
@@ -142,47 +150,82 @@ type CachingHandler struct {
 
 ```go
 type Cache struct {
-    mu      sync.RWMutex
-    entries map[string]*CacheEntry
-    ttl     time.Duration  // default: 10 * time.Second
+    mu         sync.RWMutex
+    entries    map[string]*CacheEntry
+    defaultTTL time.Duration
+    keepStale  bool
+
+    // Request coalescing for identical range requests.
+    inflight   map[string]*inflightRequest
+    inflightMu sync.Mutex
+}
+
+func RegKey(slaveID, functionCode byte, address uint16) string {
+    return fmt.Sprintf("%d:%d:%d", slaveID, functionCode, address)
 }
 
-func (c *Cache) Get(key string) ([]byte, bool) {
+func RangeKey(slaveID, functionCode byte, address, quantity uint16) string {
+    return fmt.Sprintf("%d:%d:%d:%d", slaveID, functionCode, address, quantity)
+}
+
+func (c *Cache) GetRange(slaveID, functionCode byte, address, quantity uint16) ([][]byte, bool) {
+    if quantity == 0 {
+        return nil, false
+    }
+
     c.mu.RLock()
     defer c.mu.RUnlock()
-    
-    entry, ok := c.entries[key]
-    if !ok || time.Since(entry.Timestamp) > entry.TTL {
-        return nil, false
+
+    values := make([][]byte, quantity)
+    for i := uint16(0); i < quantity; i++ {
+        entry, ok := c.entries[RegKey(slaveID, functionCode, address+i)]
+        if !ok || entry.IsExpired() {
+            return nil, false
+        }
+        values[i] = append([]byte(nil), entry.Data...)
+    }
+    return values, true
+}
+
+func (c *Cache) SetRange(slaveID, functionCode byte, address uint16, values [][]byte) {
+    c.mu.Lock()
+    defer c.mu.Unlock()
+
+    now := time.Now()
+    for i, value := range values {
+        c.entries[RegKey(slaveID, functionCode, address+uint16(i))] = &CacheEntry{
+            Data:      append([]byte(nil), value...),
+            Timestamp: now,
+            TTL:       c.defaultTTL,
+        }
     }
-    return entry.Data, true
 }
 
-func (c *Cache) Set(key string, data []byte, ttl time.Duration) {
+func (c *Cache) DeleteRange(slaveID, functionCode byte, address, quantity uint16) {
     c.mu.Lock()
     defer c.mu.Unlock()
-    
-    c.entries[key] = &CacheEntry{
-        Data:      data,
-        Timestamp: time.Now(),
-        TTL:       ttl,
+
+    for i := uint16(0); i < quantity; i++ {
+        delete(c.entries, RegKey(slaveID, functionCode, address+i))
     }
 }
 ```
 
+The cache also exposes `Coalesce(ctx, rangeKey, fetch)` for request coalescing. It does not read or write cache entries directly; the proxy performs cache lookups and stores decomposed responses.
+
 ### Request Flow
 
 1. Client sends Modbus TCP request
 2. Parse request: extract slave ID, function code, address, quantity
 3. **For reads**:
-   - Build cache key
-   - Check cache → if hit & valid, return cached data
-   - On miss: forward to upstream device
-   - Store response in cache
+   - 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
    - Return response to client
 4. **For writes**:
    - Check readonly mode
-   - If allowed: forward to upstream, optionally invalidate cache
+   - If allowed: forward to upstream, then invalidate every cached register/coil in the written address range
    - Return response
 
 ## Logging

internal/cache/cache_test.go

@@ -125,6 +125,18 @@ func TestCache_GetRange(t *testing.T) {
 	}
 }
 
+func TestCache_GetRangeZeroQuantityMiss(t *testing.T) {
+	c := New(time.Second, false)
+	defer c.Close()
+
+	if _, ok := c.GetRange(1, 0x03, 10, 0); ok {
+		t.Error("expected zero-quantity range to miss")
+	}
+	if _, ok := c.GetRangeStale(1, 0x03, 10, 0); ok {
+		t.Error("expected zero-quantity stale range to miss")
+	}
+}
+
 func TestCache_SetRange(t *testing.T) {
 	c := New(time.Second, false)
 	defer c.Close()

internal/proxy/proxy.go

@@ -13,12 +13,18 @@ import (
 	"github.com/tma/mbproxy/internal/modbus"
 )
 
+type upstreamClient interface {
+	Connect() error
+	Close() error
+	Execute(context.Context, *modbus.Request) ([]byte, error)
+}
+
 // Proxy is a caching Modbus proxy server.
 type Proxy struct {
 	cfg    *config.Config
 	logger *slog.Logger
 	server *modbus.Server
-	client *modbus.Client
+	client upstreamClient
 	cache  *cache.Cache
 }
 

internal/proxy/proxy_test.go

@@ -1,7 +1,9 @@
 package proxy
 
 import (
+	"bytes"
 	"context"
+	"errors"
 	"io"
 	"log/slog"
 	"testing"
@@ -19,6 +21,10 @@ type mockClient struct {
 	calls    int
 }
 
+func (m *mockClient) Connect() error { return nil }
+
+func (m *mockClient) Close() error { return nil }
+
 func (m *mockClient) Execute(ctx context.Context, req *modbus.Request) ([]byte, error) {
 	m.calls++
 	return m.response, m.err
@@ -69,6 +75,113 @@ func TestProxy_HandleReadCacheHit(t *testing.T) {
 	}
 }
 
+func TestProxy_HandleReadMissFetchesAndCaches(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	c := cache.New(time.Second, false)
+	defer c.Close()
+
+	upstream := &mockClient{
+		response: []byte{0x03, 0x04, 0x00, 0x0A, 0x00, 0x0B},
+	}
+	p := &Proxy{
+		cfg: &config.Config{
+			CacheTTL:        time.Second,
+			CacheServeStale: false,
+			ReadOnly:        config.ReadOnlyOn,
+		},
+		logger: logger,
+		client: upstream,
+		cache:  c,
+	}
+
+	req := &modbus.Request{
+		SlaveID:      1,
+		FunctionCode: modbus.FuncReadHoldingRegisters,
+		Address:      10,
+		Quantity:     2,
+	}
+
+	resp, err := p.HandleRequest(context.Background(), req)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	expected := []byte{0x03, 0x04, 0x00, 0x0A, 0x00, 0x0B}
+	if !bytes.Equal(resp, expected) {
+		t.Fatalf("first response: expected %v, got %v", expected, resp)
+	}
+	if upstream.calls != 1 {
+		t.Fatalf("expected 1 upstream call after miss, got %d", upstream.calls)
+	}
+
+	values, ok := c.GetRange(1, modbus.FuncReadHoldingRegisters, 10, 2)
+	if !ok {
+		t.Fatal("expected fetched response to be cached per register")
+	}
+	if !bytes.Equal(values[0], []byte{0x00, 0x0A}) || !bytes.Equal(values[1], []byte{0x00, 0x0B}) {
+		t.Fatalf("unexpected cached values: %v", values)
+	}
+
+	// Change the upstream response. The second request should be served from cache,
+	// so the upstream should not be called again and the response should stay the same.
+	upstream.response = []byte{0x03, 0x04, 0x00, 0xFF, 0x00, 0xFF}
+	resp, err = p.HandleRequest(context.Background(), req)
+	if err != nil {
+		t.Fatalf("unexpected error on cached read: %v", err)
+	}
+	if !bytes.Equal(resp, expected) {
+		t.Fatalf("cached response: expected %v, got %v", expected, resp)
+	}
+	if upstream.calls != 1 {
+		t.Fatalf("expected cached read to avoid upstream call, got %d calls", upstream.calls)
+	}
+}
+
+func TestProxy_HandleReadServesStaleOnUpstreamError(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	c := cache.New(10*time.Millisecond, true)
+	defer c.Close()
+
+	c.SetRange(1, modbus.FuncReadHoldingRegisters, 20, [][]byte{
+		{0x00, 0x01},
+		{0x00, 0x02},
+	})
+	time.Sleep(20 * time.Millisecond)
+
+	upstreamErr := errors.New("upstream unavailable")
+	upstream := &mockClient{err: upstreamErr}
+	p := &Proxy{
+		cfg: &config.Config{
+			CacheTTL:        10 * time.Millisecond,
+			CacheServeStale: true,
+			ReadOnly:        config.ReadOnlyOn,
+		},
+		logger: logger,
+		client: upstream,
+		cache:  c,
+	}
+
+	req := &modbus.Request{
+		SlaveID:      1,
+		FunctionCode: modbus.FuncReadHoldingRegisters,
+		Address:      20,
+		Quantity:     2,
+	}
+
+	resp, err := p.HandleRequest(context.Background(), req)
+	if err != nil {
+		t.Fatalf("expected stale response, got error: %v", err)
+	}
+	if upstream.calls != 1 {
+		t.Fatalf("expected one failed upstream call before serving stale, got %d", upstream.calls)
+	}
+
+	expected := []byte{0x03, 0x04, 0x00, 0x01, 0x00, 0x02}
+	if !bytes.Equal(resp, expected) {
+		t.Fatalf("stale response: expected %v, got %v", expected, resp)
+	}
+}
+
 func TestProxy_HandleWriteReadOnlyMode(t *testing.T) {
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))