diff --git a/README.md b/README.md index 043da77..242ef56 100644 --- a/README.md +++ b/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 diff --git a/SPEC.md b/SPEC.md index dc6b7bf..00fd9b9 100644 --- a/SPEC.md +++ b/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" diff --git a/internal/cache/cache.go b/internal/cache/cache.go index b9126ff..9a775b9 100644 --- a/internal/cache/cache.go +++ b/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. diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 2099ad8..453e105 100644 --- a/internal/cache/cache_test.go +++ b/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() diff --git a/internal/config/config.go b/internal/config/config.go index 7e933bf..e95b69e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -26,7 +26,8 @@ type Config struct { CacheTTL time.Duration CacheServeStale bool ReadOnly ReadOnlyMode - Timeout time.Duration + AttemptTimeout time.Duration + RequestTimeout time.Duration RequestDelay time.Duration ConnectDelay time.Duration ShutdownTimeout time.Duration @@ -43,7 +44,8 @@ func Load() (*Config, error) { CacheTTL: 10 * time.Second, CacheServeStale: false, ReadOnly: ReadOnlyOn, - Timeout: 10 * time.Second, + AttemptTimeout: 10 * time.Second, + RequestTimeout: 30 * time.Second, RequestDelay: 0, ConnectDelay: 0, ShutdownTimeout: 30 * time.Second, @@ -92,13 +94,36 @@ func Load() (*Config, error) { } } - // Parse timeout - if s := os.Getenv("MODBUS_TIMEOUT"); s != "" { + attemptTimeout, attemptTimeoutSet, err := parsePositiveDuration("MODBUS_ATTEMPT_TIMEOUT") + if err != nil { + return nil, err + } + legacyTimeout, legacyTimeoutSet, err := parsePositiveDuration("MODBUS_TIMEOUT") + if err != nil { + return nil, err + } + if attemptTimeoutSet && legacyTimeoutSet && attemptTimeout != legacyTimeout { + return nil, fmt.Errorf( + "MODBUS_ATTEMPT_TIMEOUT (%s) and deprecated MODBUS_TIMEOUT (%s) must match when both are set", + attemptTimeout, + legacyTimeout, + ) + } + if attemptTimeoutSet { + cfg.AttemptTimeout = attemptTimeout + } else if legacyTimeoutSet { + cfg.AttemptTimeout = legacyTimeout + } + + if s := os.Getenv("MODBUS_REQUEST_TIMEOUT"); s != "" { d, err := time.ParseDuration(s) if err != nil { - return nil, fmt.Errorf("invalid MODBUS_TIMEOUT: %w", err) + return nil, fmt.Errorf("invalid MODBUS_REQUEST_TIMEOUT: %w", err) + } + if d <= 0 { + return nil, fmt.Errorf("invalid MODBUS_REQUEST_TIMEOUT: must be greater than zero") } - cfg.Timeout = d + cfg.RequestTimeout = d } // Parse request delay @@ -131,6 +156,21 @@ func Load() (*Config, error) { return cfg, nil } +func parsePositiveDuration(name string) (time.Duration, bool, error) { + s := os.Getenv(name) + if s == "" { + return 0, false, nil + } + d, err := time.ParseDuration(s) + if err != nil { + return 0, false, fmt.Errorf("invalid %s: %w", name, err) + } + if d <= 0 { + return 0, false, fmt.Errorf("invalid %s: must be greater than zero", name) + } + return d, true, nil +} + // GetEnv returns the value of the environment variable named by key, // or defaultValue if the variable is not set. func GetEnv(key, defaultValue string) string { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ebc316b..5025e08 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -2,6 +2,7 @@ package config import ( "os" + "strings" "testing" "time" ) @@ -17,7 +18,9 @@ func TestLoad_Defaults(t *testing.T) { os.Unsetenv("MODBUS_CACHE_TTL") os.Unsetenv("MODBUS_CACHE_SERVE_STALE") os.Unsetenv("MODBUS_READONLY") + os.Unsetenv("MODBUS_ATTEMPT_TIMEOUT") os.Unsetenv("MODBUS_TIMEOUT") + os.Unsetenv("MODBUS_REQUEST_TIMEOUT") os.Unsetenv("MODBUS_SHUTDOWN_TIMEOUT") os.Unsetenv("HEALTH_LISTEN") os.Unsetenv("LOG_LEVEL") @@ -45,8 +48,11 @@ func TestLoad_Defaults(t *testing.T) { if cfg.ReadOnly != ReadOnlyOn { t.Errorf("expected readonly true, got %s", cfg.ReadOnly) } - if cfg.Timeout != 10*time.Second { - t.Errorf("expected 10s timeout, got %v", cfg.Timeout) + if cfg.AttemptTimeout != 10*time.Second { + t.Errorf("expected 10s attempt timeout, got %v", cfg.AttemptTimeout) + } + if cfg.RequestTimeout != 30*time.Second { + t.Errorf("expected 30s request timeout, got %v", cfg.RequestTimeout) } if cfg.RequestDelay != 0 { t.Errorf("expected 0 request delay, got %v", cfg.RequestDelay) @@ -81,7 +87,9 @@ func TestLoad_CustomValues(t *testing.T) { os.Setenv("MODBUS_CACHE_TTL", "30s") os.Setenv("MODBUS_CACHE_SERVE_STALE", "true") os.Setenv("MODBUS_READONLY", "false") - os.Setenv("MODBUS_TIMEOUT", "5s") + os.Setenv("MODBUS_ATTEMPT_TIMEOUT", "5s") + os.Unsetenv("MODBUS_TIMEOUT") + os.Setenv("MODBUS_REQUEST_TIMEOUT", "9s") os.Setenv("MODBUS_REQUEST_DELAY", "100ms") os.Setenv("MODBUS_CONNECT_DELAY", "200ms") os.Setenv("MODBUS_SHUTDOWN_TIMEOUT", "60s") @@ -94,7 +102,9 @@ func TestLoad_CustomValues(t *testing.T) { os.Unsetenv("MODBUS_CACHE_TTL") os.Unsetenv("MODBUS_CACHE_SERVE_STALE") os.Unsetenv("MODBUS_READONLY") + os.Unsetenv("MODBUS_ATTEMPT_TIMEOUT") os.Unsetenv("MODBUS_TIMEOUT") + os.Unsetenv("MODBUS_REQUEST_TIMEOUT") os.Unsetenv("MODBUS_REQUEST_DELAY") os.Unsetenv("MODBUS_CONNECT_DELAY") os.Unsetenv("MODBUS_SHUTDOWN_TIMEOUT") @@ -121,8 +131,11 @@ func TestLoad_CustomValues(t *testing.T) { if cfg.ReadOnly != ReadOnlyOff { t.Errorf("expected readonly false, got %s", cfg.ReadOnly) } - if cfg.Timeout != 5*time.Second { - t.Errorf("expected 5s timeout, got %v", cfg.Timeout) + if cfg.AttemptTimeout != 5*time.Second { + t.Errorf("expected 5s attempt timeout, got %v", cfg.AttemptTimeout) + } + if cfg.RequestTimeout != 9*time.Second { + t.Errorf("expected 9s request timeout, got %v", cfg.RequestTimeout) } if cfg.RequestDelay != 100*time.Millisecond { t.Errorf("expected 100ms request delay, got %v", cfg.RequestDelay) @@ -186,8 +199,10 @@ func TestLoad_InvalidDuration(t *testing.T) { os.Setenv("MODBUS_UPSTREAM", "localhost:502") defer os.Unsetenv("MODBUS_UPSTREAM") - tests := []string{"MODBUS_CACHE_TTL", "MODBUS_TIMEOUT", "MODBUS_REQUEST_DELAY", "MODBUS_CONNECT_DELAY", "MODBUS_SHUTDOWN_TIMEOUT"} + tests := []string{"MODBUS_CACHE_TTL", "MODBUS_ATTEMPT_TIMEOUT", "MODBUS_TIMEOUT", "MODBUS_REQUEST_TIMEOUT", "MODBUS_REQUEST_DELAY", "MODBUS_CONNECT_DELAY", "MODBUS_SHUTDOWN_TIMEOUT"} for _, envVar := range tests { + os.Unsetenv("MODBUS_ATTEMPT_TIMEOUT") + os.Unsetenv("MODBUS_TIMEOUT") os.Setenv(envVar, "invalid") _, err := Load() if err == nil { @@ -197,13 +212,70 @@ func TestLoad_InvalidDuration(t *testing.T) { } } +func TestLoad_NonPositiveTimeouts(t *testing.T) { + t.Setenv("MODBUS_UPSTREAM", "localhost:502") + for _, envVar := range []string{"MODBUS_ATTEMPT_TIMEOUT", "MODBUS_TIMEOUT", "MODBUS_REQUEST_TIMEOUT"} { + t.Run(envVar, func(t *testing.T) { + t.Setenv("MODBUS_ATTEMPT_TIMEOUT", "") + t.Setenv("MODBUS_TIMEOUT", "") + t.Setenv(envVar, "0") + if _, err := Load(); err == nil { + t.Fatalf("expected error for zero %s", envVar) + } + }) + } +} + +func TestLoad_AttemptTimeoutMigration(t *testing.T) { + tests := []struct { + name string + preferred string + legacy string + want time.Duration + errContains string + }{ + {name: "preferred only", preferred: "4s", want: 4 * time.Second}, + {name: "legacy only", legacy: "6s", want: 6 * time.Second}, + {name: "both equal after parsing", preferred: "5s", legacy: "5000ms", want: 5 * time.Second}, + {name: "both conflict", preferred: "5s", legacy: "6s", errContains: "must match when both are set"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("MODBUS_UPSTREAM", "localhost:502") + t.Setenv("MODBUS_ATTEMPT_TIMEOUT", tt.preferred) + t.Setenv("MODBUS_TIMEOUT", tt.legacy) + t.Setenv("MODBUS_REQUEST_TIMEOUT", "17s") + + cfg, err := Load() + if tt.errContains != "" { + if err == nil || !strings.Contains(err.Error(), tt.errContains) { + t.Fatalf("expected error containing %q, got %v", tt.errContains, err) + } + return + } + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.AttemptTimeout != tt.want { + t.Fatalf("expected attempt timeout %v, got %v", tt.want, cfg.AttemptTimeout) + } + if cfg.RequestTimeout != 17*time.Second { + t.Fatalf("attempt timeout changed request timeout to %v", cfg.RequestTimeout) + } + }) + } +} + func TestLoad_HealthListenCustom(t *testing.T) { // Ensure optional env vars that Load() may read do not inherit // potentially invalid values from the surrounding environment. t.Setenv("MODBUS_LISTEN", "") t.Setenv("MODBUS_READONLY", "") t.Setenv("MODBUS_CACHE_TTL", "") + t.Setenv("MODBUS_ATTEMPT_TIMEOUT", "") t.Setenv("MODBUS_TIMEOUT", "") + t.Setenv("MODBUS_REQUEST_TIMEOUT", "") t.Setenv("MODBUS_REQUEST_DELAY", "") t.Setenv("MODBUS_CONNECT_DELAY", "") t.Setenv("MODBUS_SHUTDOWN_TIMEOUT", "") diff --git a/internal/modbus/client.go b/internal/modbus/client.go index 92edd69..0265459 100644 --- a/internal/modbus/client.go +++ b/internal/modbus/client.go @@ -3,184 +3,604 @@ package modbus import ( "context" "encoding/binary" + "errors" "fmt" "log/slog" "net" "sync" "time" - "github.com/grid-x/modbus" + gridmodbus "github.com/grid-x/modbus" ) -// Client wraps a Modbus TCP client with auto-reconnect capability. -type Client struct { - address string - timeout time.Duration - requestDelay time.Duration - connectDelay time.Duration - logger *slog.Logger +type requestClient interface { + ReadCoils(context.Context, uint16, uint16) ([]byte, error) + ReadDiscreteInputs(context.Context, uint16, uint16) ([]byte, error) + ReadHoldingRegisters(context.Context, uint16, uint16) ([]byte, error) + ReadInputRegisters(context.Context, uint16, uint16) ([]byte, error) + WriteSingleCoil(context.Context, uint16, uint16) ([]byte, error) + WriteSingleRegister(context.Context, uint16, uint16) ([]byte, error) + WriteMultipleCoils(context.Context, uint16, uint16, []byte) ([]byte, error) + WriteMultipleRegisters(context.Context, uint16, uint16, []byte) ([]byte, error) +} - mu sync.Mutex - client modbus.Client - conn *modbus.TCPClientHandler +type clientSession interface { + Connect(context.Context) error + Close() error + SetSlave(byte) + BeginRequest(context.Context, time.Duration) func() error +} - healthMu sync.RWMutex - lastErr error - connected bool +type sessionFactory func() (clientSession, requestClient) + +type guardedConn struct { + net.Conn + + mu sync.Mutex + ctx context.Context + maxDeadline time.Time + canceledErr error + writeStarted bool } -// NewClient creates a new Modbus TCP client. -func NewClient(address string, timeout, requestDelay, connectDelay time.Duration, logger *slog.Logger) *Client { - return &Client{ - address: address, - timeout: timeout, - requestDelay: requestDelay, - connectDelay: connectDelay, - logger: logger, +func newGuardedConn(conn net.Conn) *guardedConn { + return &guardedConn{Conn: conn} +} + +func (c *guardedConn) bind(ctx context.Context, maxDeadline time.Time) error { + c.mu.Lock() + defer c.mu.Unlock() + + c.ctx = ctx + c.maxDeadline = maxDeadline + c.canceledErr = contextError(ctx) + c.writeStarted = false + return c.Conn.SetDeadline(c.clampDeadlineLocked(maxDeadline)) +} + +func (c *guardedConn) unbind() { + c.mu.Lock() + c.ctx = nil + c.maxDeadline = time.Time{} + c.canceledErr = nil + c.writeStarted = false + c.mu.Unlock() +} + +func (c *guardedConn) cancel(err error) error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.ctx == nil { + return nil } + c.canceledErr = err + return c.Conn.SetDeadline(time.Now()) } -// Connect establishes a connection to the upstream Modbus device. -func (c *Client) Connect() error { +func (c *guardedConn) requestErrorLocked() error { + if c.ctx == nil { + return nil + } + if c.canceledErr != nil { + return c.canceledErr + } + if err := contextError(c.ctx); err != nil { + return err + } + if !c.maxDeadline.IsZero() && !time.Now().Before(c.maxDeadline) { + return context.DeadlineExceeded + } + return nil +} + +func (c *guardedConn) clampDeadlineLocked(deadline time.Time) time.Time { + if c.requestErrorLocked() != nil { + return time.Unix(1, 0) + } + if c.ctx != nil && (deadline.IsZero() || deadline.After(c.maxDeadline)) { + return c.maxDeadline + } + return deadline +} + +func (c *guardedConn) SetDeadline(deadline time.Time) error { c.mu.Lock() defer c.mu.Unlock() + return c.Conn.SetDeadline(c.clampDeadlineLocked(deadline)) +} - return c.connectLocked(context.Background()) +func (c *guardedConn) SetReadDeadline(deadline time.Time) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.Conn.SetReadDeadline(c.clampDeadlineLocked(deadline)) } -func (c *Client) connectLocked(ctx context.Context) error { - if c.conn != nil { - c.conn.Close() +func (c *guardedConn) SetWriteDeadline(deadline time.Time) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.Conn.SetWriteDeadline(c.clampDeadlineLocked(deadline)) +} + +func (c *guardedConn) Write(data []byte) (int, error) { + c.mu.Lock() + if err := c.requestErrorLocked(); err != nil { + c.mu.Unlock() + return 0, err } + c.writeStarted = true + conn := c.Conn + c.mu.Unlock() + return conn.Write(data) +} + +type tcpSession struct { + handler *gridmodbus.TCPClientHandler + mu sync.Mutex + conn *guardedConn +} - // Custom dialer with TCP keep-alive for connection health monitoring +func newTCPSession(address string, attemptTimeout, connectDelay time.Duration) (*tcpSession, requestClient) { + session := &tcpSession{} dialer := &net.Dialer{ - Timeout: c.timeout, + Timeout: attemptTimeout, KeepAlive: 30 * time.Second, } + handler := gridmodbus.NewTCPClientHandler(address, gridmodbus.WithDialer(func(ctx context.Context, network, addr string) (net.Conn, error) { + netConn, err := dialer.DialContext(ctx, network, addr) + if err != nil { + return nil, err + } + guardedConn := newGuardedConn(netConn) + session.mu.Lock() + session.conn = guardedConn + session.mu.Unlock() + return guardedConn, nil + })) + handler.Timeout = attemptTimeout + // mbproxy owns reconnect decisions so a hidden idle reconnect cannot escape + // the active request context or retry policy. + handler.IdleTimeout = -1 + handler.ConnectDelay = connectDelay + session.handler = handler + return session, gridmodbus.NewClient(handler) +} + +func (c *tcpSession) Connect(ctx context.Context) error { + return c.handler.Connect(ctx) +} + +func (c *tcpSession) Close() error { + err := c.handler.Close() + c.mu.Lock() + c.conn = nil + c.mu.Unlock() + return err +} + +func (c *tcpSession) SetSlave(slaveID byte) { + c.handler.SetSlave(slaveID) +} + +func (c *tcpSession) BeginRequest(ctx context.Context, attemptTimeout time.Duration) func() error { + c.handler.Timeout = attemptTimeout + + c.mu.Lock() + conn := c.conn + c.mu.Unlock() + if conn == nil { + return func() error { return nil } + } + + maxDeadline := time.Now().Add(attemptTimeout) + if deadline, ok := ctx.Deadline(); ok && deadline.Before(maxDeadline) { + maxDeadline = deadline + } + bindErr := conn.bind(ctx, maxDeadline) + stop := make(chan struct{}) + stopped := make(chan error, 1) + go func() { + select { + case <-ctx.Done(): + stopped <- conn.cancel(ctx.Err()) + case <-stop: + stopped <- nil + } + }() + + return func() error { + close(stop) + watchdogErr := <-stopped + conn.unbind() + return errors.Join(bindErr, watchdogErr) + } +} + +// Client wraps a Modbus TCP client with classified retry behavior. +type Client struct { + address string + attemptTimeout time.Duration + requestDelay time.Duration + connectDelay time.Duration + logger *slog.Logger + + owner chan struct{} + session clientSession + client requestClient + newSession sessionFactory + nextRequestAt time.Time + + healthMu sync.RWMutex + lastErr error + connected bool +} + +// NewClient creates a new Modbus TCP client. +func NewClient(address string, attemptTimeout, requestDelay, connectDelay time.Duration, logger *slog.Logger) *Client { + c := &Client{ + address: address, + attemptTimeout: attemptTimeout, + requestDelay: requestDelay, + connectDelay: connectDelay, + logger: logger, + owner: make(chan struct{}, 1), + } + c.owner <- struct{}{} + c.newSession = func() (clientSession, requestClient) { + return newTCPSession(address, attemptTimeout, connectDelay) + } + return c +} - handler := modbus.NewTCPClientHandler(c.address, modbus.WithDialer(dialer.DialContext)) - handler.Timeout = c.timeout - handler.IdleTimeout = c.timeout - handler.ConnectDelay = c.connectDelay +// Connect establishes a connection to the upstream Modbus device. +func (c *Client) Connect() error { + if err := c.acquire(context.Background()); err != nil { + return err + } + defer c.release() + return c.connectLocked(context.Background()) +} - if err := handler.Connect(ctx); err != nil { +func (c *Client) connectLocked(ctx context.Context) error { + if c.session != nil { + if err := c.session.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + c.logger.Debug("closing old upstream connection failed", "error", err) + } + } + + session, client := c.newSession() + if err := session.Connect(ctx); err != nil { + wrapped := fmt.Errorf("connect to %s: %w", c.address, err) + if closeErr := session.Close(); closeErr != nil && !errors.Is(closeErr, net.ErrClosed) { + c.logger.Debug("closing failed connection attempt failed", "error", closeErr) + } c.healthMu.Lock() c.connected = false - c.lastErr = fmt.Errorf("connect to %s: %w", c.address, err) + c.lastErr = wrapped c.healthMu.Unlock() - return c.lastErr + c.session = nil + c.client = nil + return wrapped } - c.conn = handler - c.client = modbus.NewClient(handler) - + c.session = session + c.client = client c.healthMu.Lock() c.connected = true c.lastErr = nil c.healthMu.Unlock() if c.connectDelay > 0 { - c.logger.Debug("applying connect delay", "delay", c.connectDelay) + c.logger.Debug("applied connect delay", "delay", c.connectDelay) } - c.logger.Info("connected to upstream", "address", c.address) return nil } +func (c *Client) disconnectLocked() { + if c.session != nil { + if err := c.session.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + c.logger.Debug("closing failed upstream connection failed", "error", err) + } + } + c.session = nil + c.client = nil + c.healthMu.Lock() + c.connected = false + c.healthMu.Unlock() +} + // Close closes the connection to the upstream device. func (c *Client) Close() error { - c.mu.Lock() - defer c.mu.Unlock() + if err := c.acquire(context.Background()); err != nil { + return err + } + defer c.release() - if c.conn != nil { - c.conn.Close() - c.conn = nil + var err error + if c.session != nil { + err = c.session.Close() + c.session = nil c.client = nil } - c.healthMu.Lock() c.connected = false c.healthMu.Unlock() + return err +} - return nil +func (c *Client) acquire(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-c.owner: + if err := ctx.Err(); err != nil { + c.release() + return err + } + return nil + } } -// Healthy reports whether the upstream connection is healthy. -// It returns the last observed error, or nil if the last operation succeeded. +func (c *Client) release() { + c.owner <- struct{}{} +} + +// Healthy reports the last observed upstream state. func (c *Client) Healthy() error { c.healthMu.RLock() defer c.healthMu.RUnlock() - + if c.lastErr != nil { + return c.lastErr + } if !c.connected { - if c.lastErr != nil { - return c.lastErr - } return fmt.Errorf("not connected") } - return c.lastErr + return nil } -// Execute sends a Modbus request and returns the response. -// It automatically reconnects on connection failure. +// Execute sends one Modbus request, retrying a read once only after a transport failure. func (c *Client) Execute(ctx context.Context, req *Request) ([]byte, error) { - c.mu.Lock() - defer c.mu.Unlock() + if err := ValidateRequest(req); err != nil { + return nil, requestError(err, 0) + } - // Ensure connected - if c.conn == nil { - if err := c.connectLocked(ctx); err != nil { - return nil, err + if err := c.acquire(ctx); err != nil { + return nil, requestError(err, 0) + } + defer c.release() + + var lastErr error + requestSlotReady := false + for attempt := 1; attempt <= 2; attempt++ { + if c.session == nil { + err := c.connectLocked(ctx) + if err != nil { + lastErr = err + willRetry := attempt == 1 && IsReadFunction(req.FunctionCode) && isTransportError(err) && contextError(ctx) == nil + if willRetry { + c.logger.Debug("upstream connect failed, retrying read", "error", err) + continue + } + return nil, c.finalError(lastErr, attempt) + } } + + if !requestSlotReady { + if err := c.waitForRequestSlot(ctx); err != nil { + return nil, c.finalError(err, attempt-1) + } + requestSlotReady = true + } + if err := contextError(ctx); err != nil { + return nil, c.finalError(err, attempt-1) + } + c.session.SetSlave(req.SlaveID) + if err := contextError(ctx); err != nil { + return nil, c.finalError(err, attempt-1) + } + attemptTimeout := c.socketTimeoutFor(ctx) + finishRequest := c.session.BeginRequest(ctx, attemptTimeout) + if err := contextError(ctx); err != nil { + if finishErr := finishRequest(); finishErr != nil { + c.logger.Debug("finishing canceled upstream request failed", "error", finishErr) + } + return nil, c.finalError(err, attempt-1) + } + attemptStart := time.Now() + resp, err := c.executeRequest(ctx, req) + responseCompletedAt := time.Now() + attemptDuration := time.Since(attemptStart) + if finishErr := finishRequest(); finishErr != nil { + c.logger.Debug("finishing upstream request failed", "error", finishErr) + } + err = classifyRequestPathError(req, err) + kind, _ := classifyError(err) + if err != nil && kind != ErrorProtocolException { + if ctxErr := contextError(ctx); ctxErr != nil { + err = ctxErr + } + } + + if err == nil { + if c.requestDelay > 0 { + c.nextRequestAt = responseCompletedAt.Add(c.requestDelay) + } + if ctxErr := contextError(ctx); ctxErr != nil { + return nil, c.finalError(ctxErr, attempt) + } + c.recordSuccess() + c.logger.Debug("upstream request completed", + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "duration", attemptDuration, + ) + return resp, nil + } + + lastErr = err + kind, _ = classifyError(err) + willRetry := attempt == 1 && + IsReadFunction(req.FunctionCode) && + (kind == ErrorTransportTimeout || kind == ErrorTransportClosed) && + contextError(ctx) == nil + + if willRetry { + c.disconnectLocked() + c.logger.Debug("upstream request failed, retrying read", "error", err) + continue + } + if kind != ErrorProtocolException { + c.disconnectLocked() + } + return nil, c.finalError(lastErr, attempt) } - // Set slave ID - c.conn.SlaveID = req.SlaveID + return nil, c.finalError(lastErr, 2) +} - // Execute request and measure time - start := time.Now() - resp, err := c.executeRequest(ctx, req) - if err != nil { - // Try reconnect once - c.logger.Debug("upstream request failed, reconnecting", "error", err) - if reconnErr := c.connectLocked(ctx); reconnErr != nil { - return nil, fmt.Errorf("reconnect failed: %w", reconnErr) +func (c *Client) waitForRequestSlot(ctx context.Context) error { + delay := time.Until(c.nextRequestAt) + if delay <= 0 { + return contextError(ctx) + } + + c.logger.Debug("waiting for upstream request slot", "delay", delay) + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-timer.C: + return contextError(ctx) + case <-ctx.Done(): + return ctx.Err() + } +} + +func classifyRequestPathError(req *Request, err error) error { + if err == nil { + return nil + } + + var mbErr *gridmodbus.Error + if errors.As(err, &mbErr) { + if mbErr.FunctionCode == req.FunctionCode|0x80 && mbErr.ExceptionCode != 0 { + return err } - c.conn.SlaveID = req.SlaveID - start = time.Now() // Reset timer for retry - resp, err = c.executeRequest(ctx, req) - if err != nil { - c.healthMu.Lock() - c.lastErr = err - c.healthMu.Unlock() - return nil, err + return &upstreamCommunicationError{err: err} + } + + kind, _ := classifyError(err) + if kind == ErrorLocal { + return &upstreamCommunicationError{err: err} + } + return err +} + +func (c *Client) socketTimeoutFor(ctx context.Context) time.Duration { + timeout := c.attemptTimeout + if deadline, ok := ctx.Deadline(); ok { + remaining := time.Until(deadline) + if remaining < timeout { + timeout = remaining } } - duration := time.Since(start) + if timeout <= 0 { + return time.Nanosecond + } + return timeout +} +func contextError(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + if deadline, ok := ctx.Deadline(); ok && !time.Now().Before(deadline) { + return context.DeadlineExceeded + } + return nil +} + +func (c *Client) finalError(err error, attempts int) error { + reqErr := requestError(err, attempts) + if reqErr.Kind == ErrorProtocolException { + c.healthMu.Lock() + c.lastErr = nil + c.connected = true + c.healthMu.Unlock() + } else if reqErr.Kind != ErrorContextCanceled { + c.healthMu.Lock() + c.lastErr = reqErr + c.healthMu.Unlock() + } + return reqErr +} + +func (c *Client) recordSuccess() { c.healthMu.Lock() c.lastErr = nil + c.connected = true c.healthMu.Unlock() +} - c.logger.Debug("upstream request completed", - "slave_id", req.SlaveID, - "func", fmt.Sprintf("0x%02X", req.FunctionCode), - "addr", req.Address, - "qty", req.Quantity, - "duration", duration, - ) - - // Apply request delay if configured (only after successful requests) - if c.requestDelay > 0 { - c.logger.Debug("applying request delay", "delay", c.requestDelay) - select { - case <-time.After(c.requestDelay): - case <-ctx.Done(): - // Context cancelled during delay - still return the successful result +func isTransportError(err error) bool { + kind, _ := classifyError(err) + return kind == ErrorTransportTimeout || kind == ErrorTransportClosed +} + +// ValidateRequest checks local Modbus constraints without contacting upstream. +func ValidateRequest(req *Request) error { + if req == nil { + return fmt.Errorf("request is nil") + } + switch req.FunctionCode { + case FuncReadCoils, FuncReadDiscreteInputs: + if req.Quantity < 1 || req.Quantity > 2000 { + return newValidationError(ExcIllegalValue, "quantity %d must be between 1 and 2000", req.Quantity) + } + case FuncReadHoldingRegisters, FuncReadInputRegisters: + if req.Quantity < 1 || req.Quantity > 125 { + return newValidationError(ExcIllegalValue, "quantity %d must be between 1 and 125", req.Quantity) } + case FuncWriteSingleCoil: + if len(req.Data) < 2 { + return newValidationError(ExcIllegalValue, "write data requires 2 bytes") + } + value := binary.BigEndian.Uint16(req.Data) + if value != 0x0000 && value != 0xFF00 { + return newValidationError(ExcIllegalValue, "coil value 0x%04X must be 0x0000 or 0xFF00", value) + } + case FuncWriteSingleRegister: + if len(req.Data) < 2 { + return newValidationError(ExcIllegalValue, "write data requires 2 bytes") + } + case FuncWriteMultipleCoils: + if req.Quantity < 1 || req.Quantity > 1968 { + return newValidationError(ExcIllegalValue, "quantity %d must be between 1 and 1968", req.Quantity) + } + expected := int(req.Quantity+7) / 8 + if len(req.Data) != expected { + return newValidationError(ExcIllegalValue, "write data has %d bytes, expected %d", len(req.Data), expected) + } + case FuncWriteMultipleRegs: + if req.Quantity < 1 || req.Quantity > 123 { + return newValidationError(ExcIllegalValue, "quantity %d must be between 1 and 123", req.Quantity) + } + expected := int(req.Quantity) * 2 + if len(req.Data) != expected { + return newValidationError(ExcIllegalValue, "write data has %d bytes, expected %d", len(req.Data), expected) + } + default: + return newValidationError(ExcIllegalFunction, "unsupported function code: 0x%02X", req.FunctionCode) } - - return resp, nil + if uint32(req.Address)+uint32(req.Quantity) > 65536 { + return newValidationError(ExcIllegalAddress, "address range exceeds 0xFFFF") + } + return nil } func (c *Client) executeRequest(ctx context.Context, req *Request) ([]byte, error) { @@ -191,28 +611,24 @@ func (c *Client) executeRequest(ctx context.Context, req *Request) ([]byte, erro return nil, err } return c.buildReadResponse(req.FunctionCode, results), nil - case FuncReadDiscreteInputs: results, err := c.client.ReadDiscreteInputs(ctx, req.Address, req.Quantity) if err != nil { return nil, err } return c.buildReadResponse(req.FunctionCode, results), nil - case FuncReadHoldingRegisters: results, err := c.client.ReadHoldingRegisters(ctx, req.Address, req.Quantity) if err != nil { return nil, err } return c.buildReadResponse(req.FunctionCode, results), nil - case FuncReadInputRegisters: results, err := c.client.ReadInputRegisters(ctx, req.Address, req.Quantity) if err != nil { return nil, err } return c.buildReadResponse(req.FunctionCode, results), nil - case FuncWriteSingleCoil: value := binary.BigEndian.Uint16(req.Data) results, err := c.client.WriteSingleCoil(ctx, req.Address, value) @@ -220,7 +636,6 @@ func (c *Client) executeRequest(ctx context.Context, req *Request) ([]byte, erro return nil, err } return c.buildWriteResponse(req.FunctionCode, req.Address, results), nil - case FuncWriteSingleRegister: value := binary.BigEndian.Uint16(req.Data) results, err := c.client.WriteSingleRegister(ctx, req.Address, value) @@ -228,21 +643,18 @@ func (c *Client) executeRequest(ctx context.Context, req *Request) ([]byte, erro return nil, err } return c.buildWriteResponse(req.FunctionCode, req.Address, results), nil - case FuncWriteMultipleCoils: results, err := c.client.WriteMultipleCoils(ctx, req.Address, req.Quantity, req.Data) if err != nil { return nil, err } return c.buildWriteResponse(req.FunctionCode, req.Address, results), nil - case FuncWriteMultipleRegs: results, err := c.client.WriteMultipleRegisters(ctx, req.Address, req.Quantity, req.Data) if err != nil { return nil, err } return c.buildWriteResponse(req.FunctionCode, req.Address, results), nil - default: return nil, fmt.Errorf("unsupported function code: 0x%02X", req.FunctionCode) } diff --git a/internal/modbus/client_test.go b/internal/modbus/client_test.go new file mode 100644 index 0000000..00e9282 --- /dev/null +++ b/internal/modbus/client_test.go @@ -0,0 +1,966 @@ +package modbus + +import ( + "context" + "encoding/binary" + "errors" + "io" + "log/slog" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + gridmodbus "github.com/grid-x/modbus" +) + +type fakeTimeoutError struct{} + +func (fakeTimeoutError) Error() string { return "i/o timeout" } +func (fakeTimeoutError) Timeout() bool { return true } +func (fakeTimeoutError) Temporary() bool { return true } + +type fakeResult struct { + data []byte + err error + started chan struct{} + release chan struct{} + complete func() +} + +type fakeRequestClient struct { + mu sync.Mutex + results []fakeResult + calls int +} + +func (c *fakeRequestClient) next(ctx context.Context) ([]byte, error) { + c.mu.Lock() + result := c.results[c.calls] + c.calls++ + c.mu.Unlock() + if result.started != nil { + close(result.started) + } + if result.release != nil { + select { + case <-result.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if result.complete != nil { + result.complete() + } + return result.data, result.err +} + +func (c *fakeRequestClient) callCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +func (c *fakeRequestClient) ReadCoils(ctx context.Context, _, _ uint16) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) ReadDiscreteInputs(ctx context.Context, _, _ uint16) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) ReadHoldingRegisters(ctx context.Context, _, _ uint16) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) ReadInputRegisters(ctx context.Context, _, _ uint16) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) WriteSingleCoil(ctx context.Context, _, _ uint16) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) WriteSingleRegister(ctx context.Context, _, _ uint16) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) WriteMultipleCoils(ctx context.Context, _, _ uint16, _ []byte) ([]byte, error) { + return c.next(ctx) +} +func (c *fakeRequestClient) WriteMultipleRegisters(ctx context.Context, _, _ uint16, _ []byte) ([]byte, error) { + return c.next(ctx) +} + +type fakeSession struct { + connectErr error + connectHook func(context.Context) + beginHook func(context.Context) + finishErr error + connects *atomic.Int32 + closes *atomic.Int32 +} + +func (c *fakeSession) Connect(ctx context.Context) error { + c.connects.Add(1) + if c.connectHook != nil { + c.connectHook(ctx) + } + return c.connectErr +} +func (c *fakeSession) Close() error { + c.closes.Add(1) + return nil +} +func (c *fakeSession) SetSlave(byte) {} +func (c *fakeSession) BeginRequest(ctx context.Context, _ time.Duration) func() error { + if c.beginHook != nil { + c.beginHook(ctx) + } + return func() error { return c.finishErr } +} + +type fakeSessionSet struct { + client *fakeRequestClient + connectErrs []error + sessions atomic.Int32 + connects atomic.Int32 + closes atomic.Int32 + finishErr error +} + +func (s *fakeSessionSet) factory() (clientSession, requestClient) { + index := int(s.sessions.Add(1)) - 1 + var err error + if index < len(s.connectErrs) { + err = s.connectErrs[index] + } + return &fakeSession{connectErr: err, finishErr: s.finishErr, connects: &s.connects, closes: &s.closes}, s.client +} + +func newFakeClient(results []fakeResult) (*Client, *fakeSessionSet) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + client := NewClient("upstream:502", time.Second, 0, 0, logger) + sessions := &fakeSessionSet{client: &fakeRequestClient{results: results}} + client.newSession = sessions.factory + return client, sessions +} + +func readRequest() *Request { + return &Request{SlaveID: 1, FunctionCode: FuncReadHoldingRegisters, Address: 32000, Quantity: 1} +} + +func writeRequest() *Request { + return &Request{ + SlaveID: 1, + FunctionCode: FuncWriteSingleRegister, + Address: 10, + Quantity: 1, + Data: []byte{0, 1}, + } +} + +func TestClient_ProtocolExceptionIsNotRetried(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{{err: &gridmodbus.Error{ + FunctionCode: FuncReadHoldingRegisters | 0x80, + ExceptionCode: ExcIllegalAddress, + }}}) + + _, err := client.Execute(t.Context(), readRequest()) + if err == nil { + t.Fatal("expected protocol exception") + } + if sessions.client.callCount() != 1 || sessions.sessions.Load() != 1 { + t.Fatalf("protocol exception retried: calls=%d sessions=%d", sessions.client.callCount(), sessions.sessions.Load()) + } + if ErrorKindOf(err) != ErrorProtocolException { + t.Fatalf("expected protocol exception, got %s", ErrorKindOf(err)) + } + if code := DownstreamException(err); code != ExcIllegalAddress { + t.Fatalf("expected exception 0x%02X, got 0x%02X", ExcIllegalAddress, code) + } +} + +func TestClient_ReadTimeoutRetriesAfterReconnect(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {data: []byte{0x12, 0x34}}, + {data: []byte{0x56, 0x78}}, + }) + + resp, err := client.Execute(t.Context(), readRequest()) + if err != nil { + t.Fatalf("execute: %v", err) + } + if sessions.client.callCount() != 2 || sessions.sessions.Load() != 2 { + t.Fatalf("expected one retry and reconnect: calls=%d sessions=%d", sessions.client.callCount(), sessions.sessions.Load()) + } + if len(resp) != 4 || resp[2] != 0x12 || resp[3] != 0x34 { + t.Fatalf("unexpected response: % x", resp) + } + if err := client.Healthy(); err != nil { + t.Fatalf("recovered retry should remain available: %v", err) + } + + if _, err := client.Execute(t.Context(), readRequest()); err != nil { + t.Fatalf("first-attempt recovery: %v", err) + } +} + +func TestClient_SecondReadTimeoutReturnsGatewayFailure(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {err: fakeTimeoutError{}}, + }) + + _, err := client.Execute(t.Context(), readRequest()) + if err == nil { + t.Fatal("expected timeout") + } + var reqErr *RequestError + if !errors.As(err, &reqErr) { + t.Fatalf("expected RequestError, got %T", err) + } + if reqErr.Kind != ErrorTransportTimeout || reqErr.Attempts != 2 { + t.Fatalf("unexpected request error: %+v", reqErr) + } + if DownstreamException(err) != ExcGatewayTargetFailed { + t.Fatalf("expected gateway target failure, got 0x%02X", DownstreamException(err)) + } + + if sessions.client.callCount() != 2 { + t.Fatalf("expected two attempts, got %d", sessions.client.callCount()) + } +} + +func TestClient_WriteTimeoutIsNotRetried(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {data: []byte{0x00, 0x01}}, + }) + write := &Request{ + SlaveID: 1, + FunctionCode: FuncWriteSingleRegister, + Address: 10, + Quantity: 1, + Data: []byte{0x00, 0x01}, + } + + if _, err := client.Execute(t.Context(), write); err == nil { + t.Fatal("expected write timeout") + } + if sessions.client.callCount() != 1 || sessions.sessions.Load() != 1 { + t.Fatalf("write was retried: calls=%d sessions=%d", sessions.client.callCount(), sessions.sessions.Load()) + } + + if _, err := client.Execute(t.Context(), readRequest()); err != nil { + t.Fatalf("next request did not reconnect: %v", err) + } + if sessions.sessions.Load() != 2 { + t.Fatalf("expected reconnect for next request, got %d sessions", sessions.sessions.Load()) + } +} + +func TestClient_ExpiredQueueRequestNeverExecutes(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + client, sessions := newFakeClient([]fakeResult{ + {data: []byte{0x00, 0x01}, started: started, release: release}, + }) + + firstDone := make(chan error, 1) + go func() { + _, err := client.Execute(context.Background(), readRequest()) + firstDone <- err + }() + <-started + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := client.Execute(ctx, readRequest()) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected queue deadline, got %v", err) + } + if sessions.client.callCount() != 1 { + t.Fatalf("expired queued request executed: %d calls", sessions.client.callCount()) + } + + close(release) + if err := <-firstDone; err != nil { + t.Fatalf("first request: %v", err) + } + if sessions.client.callCount() != 1 { + t.Fatalf("expired request executed after ownership release: %d calls", sessions.client.callCount()) + } +} + +func TestClient_ContextDeadlineCapsAttempt(t *testing.T) { + started := make(chan struct{}) + client, _ := newFakeClient([]fakeResult{ + {started: started, release: make(chan struct{})}, + }) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + _, err := client.Execute(ctx, readRequest()) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected request deadline, got %v", err) + } + if ErrorKindOf(err) != ErrorContextDeadline { + t.Fatalf("expected context deadline classification, got %s", ErrorKindOf(err)) + } + if client.session != nil { + t.Fatal("deadline-failed connection was retained") + } +} + +func TestClient_CancellationAfterConnectPreventsAttempt(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + requests := &fakeRequestClient{results: []fakeResult{{data: []byte{0x12, 0x34}}}} + var connects atomic.Int32 + var closes atomic.Int32 + client := NewClient("upstream:502", time.Second, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + client.newSession = func() (clientSession, requestClient) { + return &fakeSession{ + connectHook: func(context.Context) { cancel() }, + connects: &connects, + closes: &closes, + }, requests + } + + _, err := client.Execute(ctx, readRequest()) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected cancellation after connect, got %v", err) + } + if requests.callCount() != 0 { + t.Fatalf("expired request executed %d wire attempts", requests.callCount()) + } +} + +func TestClient_CancellationDuringBeginRequestPreventsAttempt(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + requests := &fakeRequestClient{results: []fakeResult{{data: []byte{0x12, 0x34}}}} + var connects atomic.Int32 + var closes atomic.Int32 + client := NewClient("upstream:502", time.Second, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + client.newSession = func() (clientSession, requestClient) { + return &fakeSession{ + beginHook: func(context.Context) { cancel() }, + connects: &connects, + closes: &closes, + }, requests + } + + _, err := client.Execute(ctx, readRequest()) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected cancellation before wire attempt, got %v", err) + } + if requests.callCount() != 0 { + t.Fatalf("expired request executed %d wire attempts", requests.callCount()) + } +} + +func TestClient_SuccessRacingCancellationReturnsCancellationAndKeepsConnection(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + client, _ := newFakeClient([]fakeResult{{ + data: []byte{0x12, 0x34}, + complete: cancel, + }}) + + _, err := client.Execute(ctx, readRequest()) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected cancellation after response, got %v", err) + } + if client.session == nil { + t.Fatal("matching successful response unnecessarily closed connection") + } +} + +func TestClient_SuccessReturnsBeforePacingDelay(t *testing.T) { + tests := []struct { + name string + req *Request + result []byte + }{ + {name: "read", req: readRequest(), result: []byte{0x12, 0x34}}, + {name: "acknowledged write", req: writeRequest(), result: []byte{0, 1}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{{data: tt.result}}) + client.requestDelay = time.Second + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + if _, err := client.Execute(ctx, tt.req); err != nil { + t.Fatalf("successful request was delayed into failure: %v", err) + } + if sessions.client.callCount() != 1 { + t.Fatalf("wire calls = %d, want 1", sessions.client.callCount()) + } + }) + } +} + +func TestClient_NextRequestPacingUsesNextRequestBudget(t *testing.T) { + nextStarted := make(chan struct{}) + client, sessions := newFakeClient([]fakeResult{ + {data: []byte{0x12, 0x34}}, + {data: []byte{0x56, 0x78}, started: nextStarted}, + }) + client.requestDelay = 100 * time.Millisecond + + if _, err := client.Execute(t.Context(), readRequest()); err != nil { + t.Fatalf("first request: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if _, err := client.Execute(ctx, readRequest()); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected next request to expire during pacing, got %v", err) + } + if sessions.client.callCount() != 1 { + t.Fatalf("expired paced request made %d wire calls", sessions.client.callCount()) + } + + laterDone := make(chan error, 1) + go func() { + _, err := client.Execute(context.Background(), readRequest()) + laterDone <- err + }() + select { + case <-nextStarted: + t.Fatal("later request executed before pacing interval") + case <-time.After(10 * time.Millisecond): + } + select { + case err := <-laterDone: + if err != nil { + t.Fatalf("later request: %v", err) + } + case <-time.After(time.Second): + t.Fatal("later request did not execute after pacing interval") + } + if sessions.client.callCount() != 2 { + t.Fatalf("wire calls = %d, want 2", sessions.client.callCount()) + } +} + +func TestClient_ReadRetryIsNotPaced(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {data: []byte{0x12, 0x34}}, + }) + client.requestDelay = time.Second + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + if _, err := client.Execute(ctx, readRequest()); err != nil { + t.Fatalf("retry was incorrectly paced: %v", err) + } + if sessions.client.callCount() != 2 { + t.Fatalf("wire calls = %d, want 2", sessions.client.callCount()) + } +} + +func TestClient_FinishRequestErrorDoesNotDiscardResponse(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{{data: []byte{0x12, 0x34}}}) + sessions.finishErr = errors.New("deadline cleanup failed") + + resp, err := client.Execute(t.Context(), readRequest()) + if err != nil { + t.Fatalf("cleanup error discarded response: %v", err) + } + if len(resp) != 4 || resp[2] != 0x12 || resp[3] != 0x34 { + t.Fatalf("unexpected response: % x", resp) + } +} + +func TestClient_FinishRequestErrorDoesNotReplaceRequestError(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{{err: fakeTimeoutError{}}}) + sessions.finishErr = errors.New("deadline cleanup failed") + + _, err := client.Execute(t.Context(), writeRequest()) + if ErrorKindOf(err) != ErrorTransportTimeout { + t.Fatalf("cleanup error replaced request error: %v", err) + } +} + +func TestClient_ExpiredContextDoesNotAcquireAvailableOwnership(t *testing.T) { + client, sessions := newFakeClient(nil) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := client.Execute(ctx, readRequest()) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context cancellation, got %v", err) + } + if sessions.sessions.Load() != 0 || sessions.client.callCount() != 0 { + t.Fatalf("expired request executed: sessions=%d calls=%d", sessions.sessions.Load(), sessions.client.callCount()) + } +} + +func TestClient_CancellationWinningWithOwnershipReleaseRestoresToken(t *testing.T) { + client, _ := newFakeClient(nil) + for i := 0; i < 100; i++ { + <-client.owner + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { + result <- client.acquire(ctx) + }() + cancel() + client.release() + + if err := <-result; !errors.Is(err, context.Canceled) { + t.Fatalf("iteration %d acquired expired ownership: %v", i, err) + } + select { + case <-client.owner: + default: + t.Fatalf("iteration %d lost ownership token", i) + } + client.release() + } +} + +func TestClient_LocalValidationReturnsIllegalValueWithoutConnecting(t *testing.T) { + client, sessions := newFakeClient(nil) + req := readRequest() + req.Quantity = 0 + + _, err := client.Execute(t.Context(), req) + if ErrorKindOf(err) != ErrorLocal { + t.Fatalf("expected local error, got %s", ErrorKindOf(err)) + } + if DownstreamException(err) != ExcIllegalValue { + t.Fatalf("expected illegal value, got 0x%02X", DownstreamException(err)) + } + if sessions.sessions.Load() != 0 { + t.Fatalf("validation error connected upstream %d times", sessions.sessions.Load()) + } +} + +func TestValidateRequest_AddressRangeBoundaries(t *testing.T) { + tests := []struct { + name string + req *Request + code byte + }{ + { + name: "read final address accepted", + req: &Request{FunctionCode: FuncReadHoldingRegisters, Address: 65535, Quantity: 1}, + }, + { + name: "read one past rejected", + req: &Request{FunctionCode: FuncReadHoldingRegisters, Address: 65535, Quantity: 2}, + code: ExcIllegalAddress, + }, + { + name: "write multiple final address accepted", + req: &Request{FunctionCode: FuncWriteMultipleRegs, Address: 65535, Quantity: 1, Data: []byte{0, 1}}, + }, + { + name: "write multiple one past rejected", + req: &Request{FunctionCode: FuncWriteMultipleRegs, Address: 65535, Quantity: 2, Data: []byte{0, 1, 0, 2}}, + code: ExcIllegalAddress, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := ValidateRequest(tt.req) + if tt.code == 0 { + if err != nil { + t.Fatalf("expected valid boundary request, got %v", err) + } + return + } + if DownstreamException(err) != tt.code { + t.Fatalf("expected exception 0x%02X, got error %v", tt.code, err) + } + }) + } +} + +func TestClient_UnknownUpstreamErrorsRetryReadsAndMapToGatewayFailure(t *testing.T) { + client, sessions := newFakeClient([]fakeResult{ + {err: errors.New("modbus: transaction id mismatch")}, + {err: errors.New("modbus: response length mismatch")}, + }) + + _, err := client.Execute(t.Context(), readRequest()) + if ErrorKindOf(err) != ErrorTransportClosed { + t.Fatalf("expected transport classification, got %v", err) + } + if DownstreamException(err) != ExcGatewayTargetFailed { + t.Fatalf("expected gateway exception, got 0x%02X", DownstreamException(err)) + } + if sessions.client.callCount() != 2 || sessions.sessions.Load() != 2 { + t.Fatalf("expected reconnect and retry, calls=%d sessions=%d", sessions.client.callCount(), sessions.sessions.Load()) + } +} + +func TestClient_LoopbackProtocolException(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + serverErr := make(chan error, 1) + go func() { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + serverErr <- acceptErr + return + } + defer conn.Close() + request := make([]byte, 12) + if _, readErr := io.ReadFull(conn, request); readErr != nil { + serverErr <- readErr + return + } + response := make([]byte, 9) + copy(response[:2], request[:2]) + binary.BigEndian.PutUint16(response[4:6], 3) + response[6] = request[6] + response[7] = request[7] | 0x80 + response[8] = ExcIllegalAddress + _, writeErr := conn.Write(response) + serverErr <- writeErr + }() + + client := NewClient(listener.Addr().String(), time.Second, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(func() { + if err := client.Close(); err != nil { + t.Errorf("close client: %v", err) + } + }) + _, err = client.Execute(t.Context(), readRequest()) + if ErrorKindOf(err) != ErrorProtocolException { + t.Fatalf("expected protocol exception, got %v", err) + } + if DownstreamException(err) != ExcIllegalAddress { + t.Fatalf("expected preserved exception, got 0x%02X", DownstreamException(err)) + } + if err := <-serverErr; err != nil { + t.Fatalf("loopback server: %v", err) + } +} + +func TestClient_LoopbackMalformedModbusErrorsReconnectAndMapGateway(t *testing.T) { + tests := []struct { + name string + buildResponse func([]byte) []byte + }{ + { + name: "wrong normal function code", + buildResponse: func(request []byte) []byte { + response := make([]byte, 11) + copy(response[:2], request[:2]) + binary.BigEndian.PutUint16(response[4:6], 5) + response[6] = request[6] + response[7] = FuncReadInputRegisters + response[8] = 2 + response[9] = 0x12 + response[10] = 0x34 + return response + }, + }, + { + name: "exception missing code", + buildResponse: func(request []byte) []byte { + response := make([]byte, 8) + copy(response[:2], request[:2]) + binary.BigEndian.PutUint16(response[4:6], 2) + response[6] = request[6] + response[7] = request[7] | 0x80 + return response + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + serverErr := make(chan error, 1) + var accepts atomic.Int32 + go func() { + for range 2 { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + serverErr <- acceptErr + return + } + accepts.Add(1) + request := make([]byte, 12) + if _, readErr := io.ReadFull(conn, request); readErr != nil { + conn.Close() + serverErr <- readErr + return + } + _, writeErr := conn.Write(tt.buildResponse(request)) + closeErr := conn.Close() + if writeErr != nil { + serverErr <- writeErr + return + } + if closeErr != nil { + serverErr <- closeErr + return + } + } + serverErr <- nil + }() + + client := NewClient(listener.Addr().String(), time.Second, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(func() { + if err := client.Close(); err != nil { + t.Errorf("close client: %v", err) + } + }) + _, err = client.Execute(t.Context(), readRequest()) + if ErrorKindOf(err) != ErrorTransportClosed { + t.Fatalf("expected framing failure, got %v", err) + } + if code := DownstreamException(err); code != ExcGatewayTargetFailed { + t.Fatalf("expected gateway exception, got 0x%02X", code) + } + if accepts.Load() != 2 { + t.Fatalf("accepted %d connections, want 2", accepts.Load()) + } + if err := <-serverErr; err != nil { + t.Fatalf("loopback server: %v", err) + } + }) + } +} + +func TestClient_ContextDuringConnectDelayClosesSocket(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + socketClosed := make(chan error, 1) + go func() { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + socketClosed <- acceptErr + return + } + defer conn.Close() + var b [1]byte + _, readErr := conn.Read(b[:]) + socketClosed <- readErr + }() + + client := NewClient(listener.Addr().String(), time.Second, 0, 200*time.Millisecond, slog.New(slog.NewTextHandler(io.Discard, nil))) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + _, err = client.Execute(ctx, readRequest()) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected connect deadline, got %v", err) + } + select { + case closeErr := <-socketClosed: + if closeErr == nil { + t.Fatal("expected closed connection") + } + case <-time.After(time.Second): + t.Fatal("connection leaked after connect-delay deadline") + } +} + +func TestClient_LoopbackDeadlineClosesActiveSocket(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + socketClosed := make(chan error, 1) + go func() { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + socketClosed <- acceptErr + return + } + defer conn.Close() + request := make([]byte, 12) + if _, readErr := io.ReadFull(conn, request); readErr != nil { + socketClosed <- readErr + return + } + var b [1]byte + _, readErr := conn.Read(b[:]) + socketClosed <- readErr + }() + + client := NewClient(listener.Addr().String(), time.Second, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + start := time.Now() + _, err = client.Execute(ctx, readRequest()) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline, got %v", err) + } + if elapsed := time.Since(start); elapsed > 500*time.Millisecond { + t.Fatalf("active socket ignored request deadline: %v", elapsed) + } + select { + case closeErr := <-socketClosed: + if closeErr == nil { + t.Fatal("expected closed upstream socket") + } + case <-time.After(time.Second): + t.Fatal("upstream socket was not closed after deadline") + } +} + +func TestClient_LoopbackReadTimeoutReconnectsAndRetries(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + serverErr := make(chan error, 1) + go func() { + first, acceptErr := listener.Accept() + if acceptErr != nil { + serverErr <- acceptErr + return + } + request := make([]byte, 12) + if _, readErr := io.ReadFull(first, request); readErr != nil { + first.Close() + serverErr <- readErr + return + } + var b [1]byte + if _, readErr := first.Read(b[:]); readErr == nil { + first.Close() + serverErr <- errors.New("expected first connection to close") + return + } + if closeErr := first.Close(); closeErr != nil { + serverErr <- closeErr + return + } + + second, acceptErr := listener.Accept() + if acceptErr != nil { + serverErr <- acceptErr + return + } + defer second.Close() + if _, readErr := io.ReadFull(second, request); readErr != nil { + serverErr <- readErr + return + } + response := make([]byte, 11) + copy(response[:2], request[:2]) + binary.BigEndian.PutUint16(response[4:6], 5) + response[6] = request[6] + response[7] = request[7] + response[8] = 2 + response[9] = 0x12 + response[10] = 0x34 + _, writeErr := second.Write(response) + serverErr <- writeErr + }() + + client := NewClient(listener.Addr().String(), 20*time.Millisecond, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(func() { + if err := client.Close(); err != nil { + t.Errorf("close client: %v", err) + } + }) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + resp, err := client.Execute(ctx, readRequest()) + if err != nil { + t.Fatalf("execute: %v", err) + } + if len(resp) != 4 || resp[2] != 0x12 || resp[3] != 0x34 { + t.Fatalf("unexpected retry response: % x", resp) + } + if err := <-serverErr; err != nil { + t.Fatalf("loopback server: %v", err) + } +} + +func TestClient_LoopbackFramingErrorReconnectsAndRetries(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + serverErr := make(chan error, 1) + go func() { + request := make([]byte, 12) + first, acceptErr := listener.Accept() + if acceptErr != nil { + serverErr <- acceptErr + return + } + if _, readErr := io.ReadFull(first, request); readErr != nil { + first.Close() + serverErr <- readErr + return + } + response := make([]byte, 11) + binary.BigEndian.PutUint16(response[0:2], binary.BigEndian.Uint16(request[0:2])+1) + binary.BigEndian.PutUint16(response[4:6], 5) + response[6] = request[6] + response[7] = request[7] + response[8] = 2 + response[9] = 0xAA + response[10] = 0xBB + if _, writeErr := first.Write(response); writeErr != nil { + first.Close() + serverErr <- writeErr + return + } + if closeErr := first.Close(); closeErr != nil { + serverErr <- closeErr + return + } + + second, acceptErr := listener.Accept() + if acceptErr != nil { + serverErr <- acceptErr + return + } + defer second.Close() + if _, readErr := io.ReadFull(second, request); readErr != nil { + serverErr <- readErr + return + } + copy(response[:2], request[:2]) + response[9] = 0x12 + response[10] = 0x34 + _, writeErr := second.Write(response) + serverErr <- writeErr + }() + + client := NewClient(listener.Addr().String(), time.Second, 0, 0, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(func() { + if err := client.Close(); err != nil { + t.Errorf("close client: %v", err) + } + }) + resp, err := client.Execute(t.Context(), readRequest()) + if err != nil { + t.Fatalf("execute: %v", err) + } + if len(resp) != 4 || resp[2] != 0x12 || resp[3] != 0x34 { + t.Fatalf("unexpected retry response: % x", resp) + } + if err := <-serverErr; err != nil { + t.Fatalf("loopback server: %v", err) + } +} diff --git a/internal/modbus/connection_test.go b/internal/modbus/connection_test.go new file mode 100644 index 0000000..0a4c947 --- /dev/null +++ b/internal/modbus/connection_test.go @@ -0,0 +1,266 @@ +package modbus + +import ( + "context" + "errors" + "io" + "net" + "runtime" + "sync" + "testing" + "time" +) + +type stubAddr string + +func (a stubAddr) Network() string { return string(a) } +func (a stubAddr) String() string { return string(a) } + +type deadlineSpyConn struct { + mu sync.Mutex + writes int + lastDeadline time.Time +} + +func (c *deadlineSpyConn) Read([]byte) (int, error) { return 0, io.EOF } + +func (c *deadlineSpyConn) Write(data []byte) (int, error) { + c.mu.Lock() + c.writes++ + c.mu.Unlock() + return len(data), nil +} + +func (c *deadlineSpyConn) Close() error { return nil } +func (c *deadlineSpyConn) LocalAddr() net.Addr { return stubAddr("local") } +func (c *deadlineSpyConn) RemoteAddr() net.Addr { return stubAddr("remote") } +func (c *deadlineSpyConn) SetReadDeadline(time.Time) error { return nil } +func (c *deadlineSpyConn) SetWriteDeadline(time.Time) error { return nil } + +func (c *deadlineSpyConn) SetDeadline(deadline time.Time) error { + c.mu.Lock() + c.lastDeadline = deadline + c.mu.Unlock() + return nil +} + +func (c *deadlineSpyConn) state() (int, time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + return c.writes, c.lastDeadline +} + +func TestGuardedConn_CanceledRequestCannotWrite(t *testing.T) { + tests := []struct { + name string + context func() (context.Context, context.CancelFunc) + wantErr error + }{ + { + name: "canceled", + context: func() (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx, cancel + }, + wantErr: context.Canceled, + }, + { + name: "deadline expired", + context: func() (context.Context, context.CancelFunc) { + return context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + }, + wantErr: context.DeadlineExceeded, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + raw := &deadlineSpyConn{} + conn := newGuardedConn(raw) + ctx, cancel := tt.context() + defer cancel() + if err := conn.bind(ctx, time.Now().Add(time.Second)); err != nil { + t.Fatalf("bind: %v", err) + } + if err := conn.SetDeadline(time.Now().Add(time.Hour)); err != nil { + t.Fatalf("grid-x deadline: %v", err) + } + + n, err := conn.Write([]byte("request")) + if n != 0 || !errors.Is(err, tt.wantErr) { + t.Fatalf("write = (%d, %v), want (0, %v)", n, err, tt.wantErr) + } + writes, deadline := raw.state() + if writes != 0 { + t.Fatalf("raw writes = %d, want 0", writes) + } + if !deadline.Before(time.Now()) { + t.Fatalf("canceled deadline was extended to %v", deadline) + } + }) + } +} + +func TestGuardedConn_AbsoluteDeadlineCannotBeExtended(t *testing.T) { + raw := &deadlineSpyConn{} + conn := newGuardedConn(raw) + maxDeadline := time.Now().Add(time.Minute) + if err := conn.bind(context.Background(), maxDeadline); err != nil { + t.Fatalf("bind: %v", err) + } + if err := conn.SetDeadline(time.Now().Add(time.Hour)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + _, deadline := raw.state() + if !deadline.Equal(maxDeadline) { + t.Fatalf("raw deadline = %v, want cap %v", deadline, maxDeadline) + } +} + +func TestGuardedConn_NormalWrite(t *testing.T) { + raw := &deadlineSpyConn{} + conn := newGuardedConn(raw) + if err := conn.bind(context.Background(), time.Now().Add(time.Second)); err != nil { + t.Fatalf("bind: %v", err) + } + + n, err := conn.Write([]byte("request")) + if err != nil || n != len("request") { + t.Fatalf("write = (%d, %v)", n, err) + } + writes, _ := raw.state() + if writes != 1 { + t.Fatalf("raw writes = %d, want 1", writes) + } +} + +func TestGuardedConn_CancellationInterruptsActiveWrite(t *testing.T) { + raw, peer := net.Pipe() + defer raw.Close() + defer peer.Close() + conn := newGuardedConn(raw) + if err := conn.bind(context.Background(), time.Now().Add(time.Second)); err != nil { + t.Fatalf("bind: %v", err) + } + + writeDone := make(chan error, 1) + go func() { + _, err := conn.Write([]byte("blocked request")) + writeDone <- err + }() + + deadline := time.Now().Add(time.Second) + for { + conn.mu.Lock() + started := conn.writeStarted + conn.mu.Unlock() + if started { + break + } + if time.Now().After(deadline) { + t.Fatal("write did not start") + } + runtime.Gosched() + } + + if err := conn.cancel(context.Canceled); err != nil { + t.Fatalf("cancel: %v", err) + } + select { + case err := <-writeDone: + if err == nil { + t.Fatal("active write was not interrupted") + } + case <-time.After(time.Second): + t.Fatal("active write remained blocked after cancellation") + } +} + +func TestGuardedConn_LoopbackCanceledRequestSendsNoBytes(t *testing.T) { + tests := []struct { + name string + context func() (context.Context, context.CancelFunc) + wantErr error + }{ + { + name: "canceled", + context: func() (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx, cancel + }, + wantErr: context.Canceled, + }, + { + name: "deadline expired", + context: func() (context.Context, context.CancelFunc) { + return context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + }, + wantErr: context.DeadlineExceeded, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer listener.Close() + + type readResult struct { + n int + err error + } + serverRead := make(chan readResult, 1) + go func() { + server, acceptErr := listener.Accept() + if acceptErr != nil { + serverRead <- readResult{err: acceptErr} + return + } + defer server.Close() + if deadlineErr := server.SetReadDeadline(time.Now().Add(100 * time.Millisecond)); deadlineErr != nil { + serverRead <- readResult{err: deadlineErr} + return + } + var data [1]byte + n, readErr := server.Read(data[:]) + serverRead <- readResult{n: n, err: readErr} + }() + + raw, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer raw.Close() + conn := newGuardedConn(raw) + ctx, cancel := tt.context() + defer cancel() + if err := conn.bind(ctx, time.Now().Add(time.Second)); err != nil { + t.Fatalf("bind: %v", err) + } + if err := conn.SetDeadline(time.Now().Add(time.Hour)); err != nil { + t.Fatalf("grid-x deadline: %v", err) + } + + n, err := conn.Write([]byte("request")) + if n != 0 || !errors.Is(err, tt.wantErr) { + t.Fatalf("write = (%d, %v), want (0, %v)", n, err, tt.wantErr) + } + select { + case result := <-serverRead: + if result.n != 0 { + t.Fatalf("server received %d bytes", result.n) + } + if result.err == nil { + t.Fatal("server read unexpectedly succeeded") + } + case <-time.After(time.Second): + t.Fatal("server read did not complete") + } + }) + } +} diff --git a/internal/modbus/errors.go b/internal/modbus/errors.go new file mode 100644 index 0000000..fce33bc --- /dev/null +++ b/internal/modbus/errors.go @@ -0,0 +1,180 @@ +package modbus + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "os" + "syscall" + + gridmodbus "github.com/grid-x/modbus" +) + +// ErrorKind identifies how a request failed and whether retrying is safe. +type ErrorKind string + +const ( + ErrorProtocolException ErrorKind = "protocol_exception" + ErrorTransportTimeout ErrorKind = "transport_timeout" + ErrorTransportClosed ErrorKind = "transport_closed" + ErrorContextDeadline ErrorKind = "context_deadline" + ErrorContextCanceled ErrorKind = "context_canceled" + ErrorLocal ErrorKind = "local_error" +) + +// RequestError carries the classified request failure. +type RequestError struct { + Kind ErrorKind + ExceptionCode byte + Attempts int + Err error +} + +type validationError struct { + exceptionCode byte + err error +} + +type upstreamCommunicationError struct { + err error +} + +func (e *validationError) Error() string { + return e.err.Error() +} + +func (e *validationError) Unwrap() error { + return e.err +} + +func (e *upstreamCommunicationError) Error() string { + return e.err.Error() +} + +func (e *upstreamCommunicationError) Unwrap() error { + return e.err +} + +func (e *RequestError) Error() string { + if e.ExceptionCode != 0 { + return fmt.Sprintf("%s (exception 0x%02X): %v", e.Kind, e.ExceptionCode, e.Err) + } + return fmt.Sprintf("%s: %v", e.Kind, e.Err) +} + +// Unwrap exposes the original upstream or context error. +func (e *RequestError) Unwrap() error { + return e.Err +} + +func classifyError(err error) (ErrorKind, byte) { + if err == nil { + return "", 0 + } + if errors.Is(err, context.DeadlineExceeded) { + return ErrorContextDeadline, 0 + } + if errors.Is(err, context.Canceled) { + return ErrorContextCanceled, 0 + } + + var validationErr *validationError + if errors.As(err, &validationErr) { + return ErrorLocal, validationErr.exceptionCode + } + + var upstreamErr *upstreamCommunicationError + if errors.As(err, &upstreamErr) { + return ErrorTransportClosed, 0 + } + + var mbErr *gridmodbus.Error + if errors.As(err, &mbErr) { + return ErrorProtocolException, mbErr.ExceptionCode + } + + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return ErrorTransportTimeout, 0 + } + + if errors.Is(err, io.EOF) || + errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, net.ErrClosed) || + errors.Is(err, os.ErrClosed) || + errors.Is(err, syscall.ECONNRESET) || + errors.Is(err, syscall.ECONNABORTED) || + errors.Is(err, syscall.ECONNREFUSED) || + errors.Is(err, syscall.ENETUNREACH) || + errors.Is(err, syscall.EHOSTUNREACH) || + errors.Is(err, syscall.EPIPE) { + return ErrorTransportClosed, 0 + } + if errors.As(err, &netErr) { + return ErrorTransportClosed, 0 + } + + return ErrorLocal, 0 +} + +// ErrorKindOf returns the classified kind for an error. +func ErrorKindOf(err error) ErrorKind { + var reqErr *RequestError + if errors.As(err, &reqErr) { + return reqErr.Kind + } + kind, _ := classifyError(err) + return kind +} + +// DownstreamException maps an upstream failure to a Modbus exception code. +func DownstreamException(err error) byte { + var reqErr *RequestError + if errors.As(err, &reqErr) && reqErr.Kind == ErrorProtocolException { + return reqErr.ExceptionCode + } + if errors.As(err, &reqErr) && reqErr.Kind == ErrorLocal && reqErr.ExceptionCode != 0 { + return reqErr.ExceptionCode + } + + var upstreamErr *upstreamCommunicationError + if errors.As(err, &upstreamErr) { + return ExcGatewayTargetFailed + } + + var mbErr *gridmodbus.Error + if errors.As(err, &mbErr) { + return mbErr.ExceptionCode + } + + var validationErr *validationError + if errors.As(err, &validationErr) { + return validationErr.exceptionCode + } + + switch ErrorKindOf(err) { + case ErrorTransportTimeout, ErrorTransportClosed, ErrorContextDeadline, ErrorContextCanceled: + return ExcGatewayTargetFailed + default: + return ExcServerFailure + } +} + +func requestError(err error, attempts int) *RequestError { + kind, exceptionCode := classifyError(err) + return &RequestError{ + Kind: kind, + ExceptionCode: exceptionCode, + Attempts: attempts, + Err: err, + } +} + +func newValidationError(exceptionCode byte, format string, args ...any) error { + return &validationError{ + exceptionCode: exceptionCode, + err: fmt.Errorf(format, args...), + } +} diff --git a/internal/modbus/server.go b/internal/modbus/server.go index 15f530f..70ad507 100644 --- a/internal/modbus/server.go +++ b/internal/modbus/server.go @@ -27,10 +27,11 @@ const ( // Modbus exception codes const ( - ExcIllegalFunction = 0x01 - ExcIllegalAddress = 0x02 - ExcIllegalValue = 0x03 - ExcServerFailure = 0x04 + ExcIllegalFunction = 0x01 + ExcIllegalAddress = 0x02 + ExcIllegalValue = 0x03 + ExcServerFailure = 0x04 + ExcGatewayTargetFailed = 0x0B ) // MBAP header size (Modbus Application Protocol) @@ -56,9 +57,10 @@ type Handler interface { // Server is a Modbus TCP server. type Server struct { - listener net.Listener - handler Handler - logger *slog.Logger + listener net.Listener + handler Handler + logger *slog.Logger + requestTimeout time.Duration mu sync.Mutex conns map[net.Conn]struct{} @@ -67,11 +69,12 @@ type Server struct { } // NewServer creates a new Modbus TCP server. -func NewServer(handler Handler, logger *slog.Logger) *Server { +func NewServer(handler Handler, requestTimeout time.Duration, logger *slog.Logger) *Server { return &Server{ - handler: handler, - logger: logger, - conns: make(map[net.Conn]struct{}), + handler: handler, + logger: logger, + requestTimeout: requestTimeout, + conns: make(map[net.Conn]struct{}), } } @@ -170,12 +173,28 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) { return } - resp, err := s.handler.HandleRequest(ctx, req) + requestCtx := ctx + cancel := func() {} + if s.requestTimeout > 0 { + requestCtx, cancel = context.WithTimeout(ctx, s.requestTimeout) + } + resp, err := s.handler.HandleRequest(requestCtx, req) + if err == nil { + err = contextError(requestCtx) + } + cancel() if err != nil { - s.logger.Debug("handler error", "error", err, "func", fmt.Sprintf("0x%02X", req.FunctionCode)) - // Send exception response - excResp := s.buildExceptionResponse(req, ExcServerFailure) - s.writeResponse(conn, req.TransactionID, req.SlaveID, excResp) + exceptionCode := DownstreamException(err) + s.logger.Debug("handler error", + "error", err, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "exception", fmt.Sprintf("0x%02X", exceptionCode), + ) + excResp := s.buildExceptionResponse(req, exceptionCode) + if writeErr := s.writeResponse(conn, req.TransactionID, req.SlaveID, excResp); writeErr != nil { + s.logger.Error("write exception response error", "error", writeErr) + return + } continue } diff --git a/internal/modbus/server_test.go b/internal/modbus/server_test.go index 2d360e0..9f28e2e 100644 --- a/internal/modbus/server_test.go +++ b/internal/modbus/server_test.go @@ -8,6 +8,8 @@ import ( "net" "testing" "time" + + gridmodbus "github.com/grid-x/modbus" ) // mockHandler implements Handler for testing @@ -26,7 +28,7 @@ func TestServer_AcceptConnections(t *testing.T) { response: []byte{0x03, 0x02, 0x00, 0x01}, // Read holding registers response } - server := NewServer(handler, logger) + server := NewServer(handler, time.Second, logger) if err := server.Listen("127.0.0.1:0"); err != nil { t.Fatalf("failed to listen: %v", err) } @@ -127,7 +129,7 @@ func TestServer_ParsePDU(t *testing.T) { } logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - server := NewServer(nil, logger) + server := NewServer(nil, time.Second, logger) for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -157,6 +159,88 @@ func TestServer_ParsePDU(t *testing.T) { } } +func TestServer_PreservesUpstreamException(t *testing.T) { + handler := &mockHandler{err: &RequestError{ + Kind: ErrorProtocolException, + ExceptionCode: ExcIllegalAddress, + Attempts: 1, + Err: &gridmodbus.Error{ + FunctionCode: FuncReadHoldingRegisters | 0x80, + ExceptionCode: ExcIllegalAddress, + }, + }} + resp := executeServerRequest(t, handler, time.Second) + if resp[7] != FuncReadHoldingRegisters|0x80 || resp[8] != ExcIllegalAddress { + t.Fatalf("unexpected exception response: % x", resp[7:9]) + } +} + +func TestServer_MapsRequestDeadlineToGatewayException(t *testing.T) { + handler := HandlerFunc(func(ctx context.Context, req *Request) ([]byte, error) { + <-ctx.Done() + return nil, ctx.Err() + }) + resp := executeServerRequest(t, handler, 20*time.Millisecond) + if resp[7] != FuncReadHoldingRegisters|0x80 || resp[8] != ExcGatewayTargetFailed { + t.Fatalf("unexpected exception response: % x", resp[7:9]) + } +} + +func TestServer_RejectsNominalSuccessAfterRequestDeadline(t *testing.T) { + handler := HandlerFunc(func(context.Context, *Request) ([]byte, error) { + time.Sleep(30 * time.Millisecond) + return []byte{FuncReadHoldingRegisters, 2, 0, 1}, nil + }) + resp := executeServerRequest(t, handler, 10*time.Millisecond) + if resp[7] != FuncReadHoldingRegisters|0x80 || resp[8] != ExcGatewayTargetFailed { + t.Fatalf("unexpected deadline response: % x", resp[7:9]) + } +} + +type HandlerFunc func(context.Context, *Request) ([]byte, error) + +func (f HandlerFunc) HandleRequest(ctx context.Context, req *Request) ([]byte, error) { + return f(ctx, req) +} + +func executeServerRequest(t *testing.T, handler Handler, timeout time.Duration) []byte { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + server := NewServer(handler, timeout, logger) + if err := server.Listen("127.0.0.1:0"); err != nil { + t.Fatalf("listen: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { + if err := server.Serve(ctx); err != nil { + t.Errorf("serve: %v", err) + } + }() + t.Cleanup(func() { + if err := server.Close(); err != nil { + t.Errorf("close server: %v", err) + } + }) + + conn, err := net.DialTimeout("tcp", server.listener.Addr().String(), time.Second) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if _, err := conn.Write(buildMBAPRequest(1, 1, FuncReadHoldingRegisters, 10, 2)); err != nil { + t.Fatalf("write request: %v", err) + } + if err := conn.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + resp := make([]byte, 9) + if _, err := io.ReadFull(conn, resp); err != nil { + t.Fatalf("read response: %v", err) + } + return resp +} + func TestIsWriteFunction(t *testing.T) { writes := []byte{FuncWriteSingleCoil, FuncWriteSingleRegister, FuncWriteMultipleCoils, FuncWriteMultipleRegs} for _, fc := range writes { diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 505eadd..7cf857f 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -6,6 +6,7 @@ import ( "encoding/binary" "fmt" "log/slog" + "sync" "time" "github.com/tma/mbproxy/internal/cache" @@ -27,18 +28,38 @@ type Proxy struct { server *modbus.Server client upstreamClient cache *cache.Cache + + cacheStateMu sync.Mutex + writeGeneration uint64 } // New creates a new proxy instance. func New(cfg *config.Config, logger *slog.Logger) (*Proxy, error) { + retryBudget := saturatingDurationSum( + cfg.AttemptTimeout, + cfg.AttemptTimeout, + cfg.ConnectDelay, + cfg.ConnectDelay, + cfg.RequestDelay, + ) + if cfg.RequestTimeout <= retryBudget { + logger.Warn("request budget may expire before a full read retry", + "attempt_timeout", cfg.AttemptTimeout, + "connect_delay", cfg.ConnectDelay, + "request_delay", cfg.RequestDelay, + "request_timeout", cfg.RequestTimeout, + "full_retry_budget", retryBudget, + ) + } + p := &Proxy{ cfg: cfg, logger: logger, - client: modbus.NewClient(cfg.Upstream, cfg.Timeout, cfg.RequestDelay, cfg.ConnectDelay, logger), + client: modbus.NewClient(cfg.Upstream, cfg.AttemptTimeout, cfg.RequestDelay, cfg.ConnectDelay, logger), cache: cache.New(cfg.CacheTTL, cfg.CacheServeStale), } - p.server = modbus.NewServer(p, logger) + p.server = modbus.NewServer(p, cfg.RequestTimeout, logger) return p, nil } @@ -55,6 +76,7 @@ func (p *Proxy) Run(ctx context.Context) error { "upstream", p.cfg.Upstream, "readonly", p.cfg.ReadOnly, "cache_ttl", p.cfg.CacheTTL, + "request_timeout", p.cfg.RequestTimeout, ) if err := p.server.Listen(p.cfg.Listen); err != nil { @@ -99,6 +121,13 @@ func (p *Proxy) Shutdown(timeout time.Duration) error { // HandleRequest implements modbus.Handler interface. func (p *Proxy) HandleRequest(ctx context.Context, req *modbus.Request) ([]byte, error) { + if err := modbus.ValidateRequest(req); err != nil { + if req == nil { + return nil, err + } + return modbus.BuildExceptionResponse(req.FunctionCode, modbus.DownstreamException(err)), nil + } + if modbus.IsWriteFunction(req.FunctionCode) { return p.handleWrite(ctx, req) } @@ -107,18 +136,14 @@ func (p *Proxy) HandleRequest(ctx context.Context, req *modbus.Request) ([]byte, return p.handleRead(ctx, req) } - // Unknown function code - p.logger.Debug("unknown function code", - "func", fmt.Sprintf("0x%02X", req.FunctionCode), - "slave_id", req.SlaveID, - ) - return modbus.BuildExceptionResponse(req.FunctionCode, modbus.ExcIllegalFunction), nil + return nil, fmt.Errorf("validated unsupported function code: 0x%02X", req.FunctionCode) } func (p *Proxy) handleRead(ctx context.Context, req *modbus.Request) ([]byte, error) { - // Check per-register cache + p.cacheStateMu.Lock() values, cacheHit := p.cache.GetRange(req.SlaveID, req.FunctionCode, req.Address, req.Quantity) if cacheHit { + p.cacheStateMu.Unlock() p.logger.Debug("cache hit", "slave_id", req.SlaveID, "func", fmt.Sprintf("0x%02X", req.FunctionCode), @@ -127,6 +152,8 @@ func (p *Proxy) handleRead(ctx context.Context, req *modbus.Request) ([]byte, er ) return assembleResponse(req.FunctionCode, req.Quantity, values), nil } + generation := p.writeGeneration + p.cacheStateMu.Unlock() // Cache miss — fetch with coalescing p.logger.Debug("cache miss", @@ -136,15 +163,30 @@ func (p *Proxy) handleRead(ctx context.Context, req *modbus.Request) ([]byte, er "qty", req.Quantity, ) - rangeKey := cache.RangeKey(req.SlaveID, req.FunctionCode, req.Address, req.Quantity) + rangeKey := fmt.Sprintf("%d:%s", generation, cache.RangeKey(req.SlaveID, req.FunctionCode, req.Address, req.Quantity)) data, err := p.cache.Coalesce(ctx, rangeKey, func(ctx context.Context) ([]byte, error) { - return p.client.Execute(ctx, req) + data, err := p.client.Execute(ctx, req) + if err != nil { + return nil, err + } + regValues := decomposeResponse(req.FunctionCode, req.Quantity, data) + if regValues != nil { + p.cacheStateMu.Lock() + if p.writeGeneration == generation { + p.cache.SetRange(req.SlaveID, req.FunctionCode, req.Address, regValues) + } + p.cacheStateMu.Unlock() + } + return data, nil }) if err != nil { // Try serving stale data if configured - if p.cfg.CacheServeStale { - if staleValues, ok := p.cache.GetRangeStale(req.SlaveID, req.FunctionCode, req.Address, req.Quantity); ok { + if p.cfg.CacheServeStale && modbus.ErrorKindOf(err) != modbus.ErrorProtocolException { + p.cacheStateMu.Lock() + staleValues, ok := p.cache.GetRangeStale(req.SlaveID, req.FunctionCode, req.Address, req.Quantity) + p.cacheStateMu.Unlock() + if ok { p.logger.Warn("upstream error, serving stale", "slave_id", req.SlaveID, "error", err, @@ -155,12 +197,6 @@ func (p *Proxy) handleRead(ctx context.Context, req *modbus.Request) ([]byte, er return nil, err } - // Decompose response and store per-register - regValues := decomposeResponse(req.FunctionCode, req.Quantity, data) - if regValues != nil { - p.cache.SetRange(req.SlaveID, req.FunctionCode, req.Address, regValues) - } - return data, nil } @@ -186,20 +222,41 @@ func (p *Proxy) handleWrite(ctx context.Context, req *modbus.Request) ([]byte, e case config.ReadOnlyOff: // Forward to upstream - resp, err := p.client.Execute(ctx, req) - if err != nil { - return nil, err - } - - // Invalidate per-register cache entries for the written range + // Invalidate before sending because a transport failure leaves the write + // outcome unknown and cached pre-write values are unsafe for reconciliation. + p.cacheStateMu.Lock() + p.writeGeneration++ p.invalidateCache(req) + p.cacheStateMu.Unlock() - return resp, nil + defer func() { + p.cacheStateMu.Lock() + p.writeGeneration++ + p.invalidateCache(req) + p.cacheStateMu.Unlock() + }() + + return p.client.Execute(ctx, req) } return nil, fmt.Errorf("unknown readonly mode: %s", p.cfg.ReadOnly) } +func saturatingDurationSum(durations ...time.Duration) time.Duration { + const maxDuration = time.Duration(1<<63 - 1) + var total time.Duration + for _, duration := range durations { + if duration <= 0 { + continue + } + if duration > maxDuration-total { + return maxDuration + } + total += duration + } + return total +} + func (p *Proxy) invalidateCache(req *modbus.Request) { // Invalidate per-register entries for all read function codes readFuncs := []byte{ diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index e643aec..a46f4a1 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -6,6 +6,7 @@ import ( "errors" "io" "log/slog" + "sync" "testing" "time" @@ -21,6 +22,75 @@ type mockClient struct { calls int } +type interleavingClient struct { + mu sync.Mutex + executeMu sync.Mutex + readCalls int + firstReadStarted chan struct{} + releaseFirstRead chan struct{} + writeQueued chan struct{} + writeStarted chan struct{} + writeCompleted chan struct{} + secondReadQueued chan struct{} +} + +type writeWindowClient struct { + writeEntered chan struct{} + releaseWrite chan struct{} + readExecuted chan struct{} + writeErr error +} + +func (c *interleavingClient) Connect() error { return nil } +func (c *interleavingClient) Close() error { return nil } +func (c *interleavingClient) Healthy() error { return nil } + +func (c *interleavingClient) Execute(_ context.Context, req *modbus.Request) ([]byte, error) { + if modbus.IsWriteFunction(req.FunctionCode) { + close(c.writeQueued) + c.executeMu.Lock() + defer c.executeMu.Unlock() + close(c.writeStarted) + close(c.writeCompleted) + return []byte{req.FunctionCode, byte(req.Address >> 8), byte(req.Address), 0, 2}, nil + } + + c.mu.Lock() + c.readCalls++ + call := c.readCalls + c.mu.Unlock() + if call == 1 { + c.executeMu.Lock() + defer c.executeMu.Unlock() + close(c.firstReadStarted) + <-c.releaseFirstRead + return []byte{req.FunctionCode, 2, 0, 1}, nil + } + close(c.secondReadQueued) + <-c.writeCompleted + c.executeMu.Lock() + defer c.executeMu.Unlock() + return []byte{req.FunctionCode, 2, 0, 2}, nil +} + +func (c *writeWindowClient) Connect() error { return nil } +func (c *writeWindowClient) Close() error { return nil } +func (c *writeWindowClient) Healthy() error { return nil } + +func (c *writeWindowClient) Execute(_ context.Context, req *modbus.Request) ([]byte, error) { + if modbus.IsWriteFunction(req.FunctionCode) { + close(c.writeEntered) + <-c.releaseWrite + if c.writeErr != nil { + return nil, c.writeErr + } + return []byte{req.FunctionCode, byte(req.Address >> 8), byte(req.Address), 0, 2}, nil + } + + close(c.readExecuted) + return []byte{req.FunctionCode, 2, 0, 1}, nil +} + func (m *mockClient) Connect() error { return nil } func (m *mockClient) Close() error { return nil } @@ -184,6 +254,39 @@ func TestProxy_HandleReadServesStaleOnUpstreamError(t *testing.T) { } } +func TestProxy_HandleReadDoesNotHideProtocolExceptionWithStaleData(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + c := cache.New(time.Millisecond, true) + defer c.Close() + c.SetRange(1, modbus.FuncReadHoldingRegisters, 20, [][]byte{{0x00, 0x01}}) + time.Sleep(2 * time.Millisecond) + + upstreamErr := &modbus.RequestError{ + Kind: modbus.ErrorProtocolException, + ExceptionCode: modbus.ExcIllegalAddress, + Err: errors.New("upstream exception"), + } + p := &Proxy{ + cfg: &config.Config{ + CacheServeStale: true, + ReadOnly: config.ReadOnlyOn, + }, + logger: logger, + client: &mockClient{err: upstreamErr}, + cache: c, + } + req := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 20, + Quantity: 1, + } + + if _, err := p.HandleRequest(t.Context(), req); !errors.Is(err, upstreamErr) { + t.Fatalf("expected protocol exception, got %v", err) + } +} + func TestProxy_HandleWriteReadOnlyMode(t *testing.T) { logger := slog.New(slog.NewTextHandler(io.Discard, nil)) @@ -264,6 +367,85 @@ func TestProxy_HandleUnknownFunction(t *testing.T) { } } +func TestProxy_InvalidReadQuantityReturnsIllegalValueWithoutUpstream(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + c := cache.New(time.Second, false) + defer c.Close() + upstream := &mockClient{} + p := &Proxy{ + cfg: &config.Config{ReadOnly: config.ReadOnlyOn}, + logger: logger, + client: upstream, + cache: c, + } + req := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 0, + Quantity: 0, + } + + resp, err := p.HandleRequest(t.Context(), req) + if err != nil { + t.Fatalf("handle request: %v", err) + } + if !bytes.Equal(resp, []byte{modbus.FuncReadHoldingRegisters | 0x80, modbus.ExcIllegalValue}) { + t.Fatalf("unexpected validation response: % x", resp) + } + if upstream.calls != 0 { + t.Fatalf("invalid request reached upstream %d times", upstream.calls) + } +} + +func TestProxy_AddressRangeOverflowDoesNotReachUpstream(t *testing.T) { + tests := []struct { + name string + req *modbus.Request + }{ + { + name: "read", + req: &modbus.Request{ + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 65535, + Quantity: 2, + }, + }, + { + name: "write multiple", + req: &modbus.Request{ + FunctionCode: modbus.FuncWriteMultipleRegs, + Address: 65535, + Quantity: 2, + Data: []byte{0, 1, 0, 2}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := cache.New(time.Second, false) + defer c.Close() + upstream := &mockClient{} + p := &Proxy{ + cfg: &config.Config{ReadOnly: config.ReadOnlyOff}, + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + client: upstream, + cache: c, + } + + resp, err := p.HandleRequest(t.Context(), tt.req) + if err != nil { + t.Fatalf("handle request: %v", err) + } + if !bytes.Equal(resp, []byte{tt.req.FunctionCode | 0x80, modbus.ExcIllegalAddress}) { + t.Fatalf("unexpected validation response: % x", resp) + } + if upstream.calls != 0 { + t.Fatalf("invalid range reached upstream %d times", upstream.calls) + } + }) + } +} + func TestProxy_BuildFakeWriteResponse(t *testing.T) { p := &Proxy{} @@ -524,3 +706,218 @@ func TestProxy_WriteInvalidatesMultipleRegisters(t *testing.T) { } } } + +func TestProxy_AmbiguousWriteInvalidatesCache(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + c := cache.New(time.Second, false) + defer c.Close() + c.SetRange(1, modbus.FuncReadHoldingRegisters, 5, [][]byte{{0x00, 0x01}}) + + upstreamErr := &modbus.RequestError{ + Kind: modbus.ErrorTransportTimeout, + Attempts: 1, + Err: errors.New("timeout"), + } + p := &Proxy{ + cfg: &config.Config{ReadOnly: config.ReadOnlyOff}, + logger: logger, + client: &mockClient{err: upstreamErr}, + cache: c, + } + req := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncWriteSingleRegister, + Address: 5, + Quantity: 1, + Data: []byte{0x00, 0x02}, + } + + if _, err := p.HandleRequest(t.Context(), req); !errors.Is(err, upstreamErr) { + t.Fatalf("expected ambiguous write error, got %v", err) + } + if _, ok := c.Get(cache.RegKey(1, modbus.FuncReadHoldingRegisters, 5)); ok { + t.Fatal("ambiguous write retained pre-write cache value") + } +} + +func TestProxy_WriteGenerationPreventsStaleReadRepopulationAndCoalescing(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + c := cache.New(time.Second, false) + defer c.Close() + upstream := &interleavingClient{ + firstReadStarted: make(chan struct{}), + releaseFirstRead: make(chan struct{}), + writeQueued: make(chan struct{}), + writeStarted: make(chan struct{}), + writeCompleted: make(chan struct{}), + secondReadQueued: make(chan struct{}), + } + p := &Proxy{ + cfg: &config.Config{ReadOnly: config.ReadOnlyOff}, + logger: logger, + client: upstream, + cache: c, + } + read := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 5, + Quantity: 1, + } + write := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncWriteSingleRegister, + Address: 5, + Quantity: 1, + Data: []byte{0, 2}, + } + + firstDone := make(chan error, 1) + go func() { + _, err := p.HandleRequest(context.Background(), read) + firstDone <- err + }() + <-upstream.firstReadStarted + + writeDone := make(chan error, 1) + go func() { + _, err := p.HandleRequest(context.Background(), write) + writeDone <- err + }() + <-upstream.writeQueued + + secondDone := make(chan struct{}) + var secondResp []byte + var secondErr error + go func() { + secondResp, secondErr = p.HandleRequest(context.Background(), read) + close(secondDone) + }() + select { + case <-upstream.secondReadQueued: + case <-time.After(time.Second): + t.Fatal("post-write read joined pre-write coalesced fetch") + } + select { + case <-secondDone: + t.Fatal("post-write read bypassed serialized write") + default: + } + + close(upstream.releaseFirstRead) + select { + case <-upstream.writeStarted: + case <-time.After(time.Second): + t.Fatal("write did not follow pre-write read") + } + if err := <-writeDone; err != nil { + t.Fatalf("write: %v", err) + } + select { + case <-secondDone: + case <-time.After(time.Second): + t.Fatal("post-write read did not complete after write") + } + if secondErr != nil || !bytes.Equal(secondResp, []byte{modbus.FuncReadHoldingRegisters, 2, 0, 2}) { + t.Fatalf("unexpected post-write response % x, err=%v", secondResp, secondErr) + } + + if err := <-firstDone; err != nil { + t.Fatalf("pre-write read: %v", err) + } + values, ok := c.GetRange(1, modbus.FuncReadHoldingRegisters, 5, 1) + if ok && !bytes.Equal(values[0], []byte{0, 2}) { + t.Fatalf("pre-write read repopulated stale cache: %v, present=%v", values, ok) + } +} + +func TestProxy_PostWriteInvalidationClosesSchedulingWindow(t *testing.T) { + tests := []struct { + name string + err error + }{ + {name: "success"}, + {name: "protocol error", err: &modbus.RequestError{ + Kind: modbus.ErrorProtocolException, + ExceptionCode: modbus.ExcIllegalAddress, + Attempts: 1, + Err: errors.New("protocol exception"), + }}, + {name: "transport error", err: &modbus.RequestError{ + Kind: modbus.ErrorTransportClosed, + Attempts: 1, + Err: errors.New("connection closed"), + }}, + {name: "context error", err: context.DeadlineExceeded}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + c := cache.New(time.Second, false) + defer c.Close() + upstream := &writeWindowClient{ + writeEntered: make(chan struct{}), + releaseWrite: make(chan struct{}), + readExecuted: make(chan struct{}), + writeErr: tt.err, + } + p := &Proxy{ + cfg: &config.Config{ReadOnly: config.ReadOnlyOff}, + logger: logger, + client: upstream, + cache: c, + } + read := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 5, + Quantity: 1, + } + write := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncWriteSingleRegister, + Address: 5, + Quantity: 1, + Data: []byte{0, 2}, + } + + writeDone := make(chan error, 1) + go func() { + _, err := p.HandleRequest(context.Background(), write) + writeDone <- err + }() + <-upstream.writeEntered + + if _, err := p.HandleRequest(context.Background(), read); err != nil { + t.Fatalf("read in scheduling window: %v", err) + } + <-upstream.readExecuted + if values, ok := c.GetRange(1, modbus.FuncReadHoldingRegisters, 5, 1); !ok || !bytes.Equal(values[0], []byte{0, 1}) { + t.Fatalf("read did not cache pre-write value: %v, present=%v", values, ok) + } + + close(upstream.releaseWrite) + err := <-writeDone + if !errors.Is(err, tt.err) { + t.Fatalf("write error = %v, want %v", err, tt.err) + } + if _, ok := c.GetRange(1, modbus.FuncReadHoldingRegisters, 5, 1); ok { + t.Fatal("post-write cleanup retained value read during scheduling window") + } + }) + } +} + +func TestSaturatingDurationSum(t *testing.T) { + const maxDuration = time.Duration(1<<63 - 1) + if got := saturatingDurationSum(maxDuration-1, 2); got != maxDuration { + t.Fatalf("expected saturated duration, got %v", got) + } + if got := saturatingDurationSum(time.Second, 2*time.Second); got != 3*time.Second { + t.Fatalf("expected normal sum, got %v", got) + } + if got := saturatingDurationSum(-time.Second, time.Second); got != time.Second { + t.Fatalf("expected negative duration to be ignored, got %v", got) + } +}