From 1a408e2076f9b75e8f93d951d022c05be18e2d71 Mon Sep 17 00:00:00 2001 From: Thomas Maurer Date: Wed, 29 Jul 2026 19:48:18 +0200 Subject: [PATCH] feat: add request diagnostics Add structured lifecycle timing, retry counters, coalescing attribution, and health degradation state without changing request safety behavior. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 97962790-10e4-4542-9e9e-02cadba19ef9 --- README.md | 30 ++++ SPEC.md | 32 +++- internal/cache/cache.go | 77 ++++++++- internal/cache/cache_test.go | 160 +++++++++++++++++- internal/health/health.go | 29 +++- internal/health/health_test.go | 110 ++++++++++++ internal/modbus/client.go | 284 +++++++++++++++++++++++++++---- internal/modbus/client_test.go | 297 ++++++++++++++++++++++++++++++++- internal/modbus/errors.go | 36 ++-- internal/modbus/server.go | 64 ++++++- internal/modbus/server_test.go | 218 ++++++++++++++++++++++++ internal/proxy/proxy.go | 129 +++++++++++++- internal/proxy/proxy_test.go | 208 ++++++++++++++++++++++- 13 files changed, 1609 insertions(+), 65 deletions(-) diff --git a/README.md b/README.md index 242ef56..aef28fd 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ A lightweight Modbus TCP proxy with in-memory caching. Designed to reduce load o - **Read-only mode**: Optionally block or ignore write requests - **Auto-reconnect**: Automatic upstream reconnection on failure - **Stale data fallback**: Optionally serve stale cache on upstream errors +- **Request diagnostics**: Structured lifecycle timing, retry, exception, and health state - **Graceful shutdown**: Complete in-flight requests before terminating - **Minimal footprint**: ~6MB Docker image (scratch base) @@ -57,6 +58,35 @@ 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. +### Production Diagnostics + +Upstream lifecycle logs identify requests with `slave_id`, `func`, `addr`, +`qty`, and `write`; write payloads are never logged. Timing fields separate +`queue_duration`, `attempt_duration`, `reconnect_duration`, and +`total_duration`. Retry and failure records also include `attempt`, `attempts`, +`error_kind`, `will_retry`, and the preserved `exception_code` or mapped +`downstream_exception` when applicable. + +The first retryable read transport failure is logged at DEBUG with +`will_retry=true`; a recovered second attempt is logged at DEBUG with +`attempts=2`. Final upstream failures and genuine upstream Modbus exceptions +are WARN records. Request cancellation is DEBUG. Requests already canceled or +expired do not select stale fallback, and the fallback record does not claim +downstream delivery. Coalesced followers report `coalesced=true`, +`coalesced_wait_duration`, and zero attempts so they are not mistaken for +independent upstream traffic. A follower that outlives a canceled leader and +performs the replacement fetch instead reports `coalesced_waited=true` with +`coalesced=false`, preserving both its wait and its real upstream attempt. + +The health response includes cumulative `total_retries` and +`recovered_retries`, consecutive first-attempt and final-failure counts, the +last first-attempt success, the last successful request, and degradation flags. +A recovered retry immediately sets `degraded=true` but remains HTTP 200 with +status `ok`. If no first-attempt success clears that state for one minute, +status becomes `degraded` while remaining HTTP 200. A final upstream failure +still returns HTTP 503 `unhealthy`. Success timestamps are `null` until the +corresponding success has occurred. + ### Read-Only Modes - `false`: Full read/write passthrough to upstream device diff --git a/SPEC.md b/SPEC.md index 00fd9b9..19a88e7 100644 --- a/SPEC.md +++ b/SPEC.md @@ -91,6 +91,34 @@ type CacheEntry struct { - Not reapplied between a failed read attempt and its retry - Logged at DEBUG level when a request waits for its slot +### Request Diagnostics and Health + +- Structured upstream lifecycle logs include request identity (`slave_id`, + `func`, `addr`, `qty`, `write`), attempt state (`attempt`, `attempts`, + `will_retry`, `error_kind`), and disjoint timing + (`queue_duration`, `attempt_duration`, `reconnect_duration`, + `total_duration`). Write payloads are excluded. +- A first retryable read transport failure is DEBUG; a successful retry is + DEBUG with `attempts=2`; a final upstream failure or genuine upstream Modbus + exception is WARN. Request cancellation is DEBUG. Requests already canceled + or expired do not select stale fallback, and the fallback record does not + claim downstream delivery. Genuine exceptions preserve their nonzero + `exception_code` and are not retried or reconnected. +- Downstream failure logs include the mapped `downstream_exception`. + Coalesced followers use `coalesced=true`, expose their wait duration, and + report zero attempts instead of inheriting the leader's wire-attempt timing. + A follower that takes over after a canceled leader instead uses + `coalesced_waited=true` with `coalesced=false`; its wait is logged separately + and its replacement fetch keeps its actual upstream-attempt diagnostics. +- Diagnostics retain cumulative retry and recovered-retry counters, + consecutive first-attempt and final-failure counts, last first-attempt + success, and last successful request. +- A recovered retry sets diagnostic degradation immediately and that state is + cleared only by a later first-attempt success. It remains available and does + not fail health. If degradation persists for one minute, the health JSON + status becomes `degraded` while HTTP status remains 200. A final upstream + failure remains `unhealthy` with HTTP 503. + ### 4. Read-Only Mode Three modes: - `false`: Full read/write passthrough @@ -259,10 +287,10 @@ The cache also exposes `Coalesce(ctx, rangeKey, fetch)` for request coalescing. 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="upstream request completed" slave_id=1 func=0x03 addr=0 qty=10 write=false attempt=1 attempts=1 queue_duration=2ms attempt_duration=15ms reconnect_duration=0s total_duration=17ms error_kind="" will_retry=false 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=WARN msg="stale fallback selected" slave_id=1 stale_fallback_selected=true error_kind=transport_timeout level=INFO msg="shutting down" ``` diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 9a775b9..9eafb76 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -43,6 +43,43 @@ type inflightRequest struct { followers int } +// CoalesceResult describes whether a caller shared another caller's fetch. +type CoalesceResult struct { + Coalesced bool + Waited bool + WaitDuration time.Duration +} + +type coalescingError struct { + err error + coalesced bool + waitDuration time.Duration +} + +func (e *coalescingError) Error() string { + return e.err.Error() +} + +func (e *coalescingError) Unwrap() error { + return e.err +} + +func (e *coalescingError) Coalesced() bool { + return e.coalesced +} + +func (e *coalescingError) CoalescedWaitDuration() time.Duration { + return e.waitDuration +} + +func attributedCoalescingError(err error, coalesced bool, waitDuration time.Duration) error { + return &coalescingError{ + err: err, + coalesced: coalesced, + waitDuration: waitDuration, + } +} + // New creates a new cache with the specified default TTL. // If keepStale is true, expired entries are retained for stale serving. func New(defaultTTL time.Duration, keepStale bool) *Cache { @@ -213,9 +250,20 @@ 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) { + data, _, err := c.CoalesceWithStatus(ctx, key, fetch) + return data, err +} + +// CoalesceWithStatus is Coalesce with explicit leader/follower attribution. +func (c *Cache) CoalesceWithStatus(ctx context.Context, key string, fetch func(context.Context) ([]byte, error)) ([]byte, CoalesceResult, error) { + var totalWait time.Duration + waited := false for { if err := ctx.Err(); err != nil { - return nil, err + if waited { + err = attributedCoalescingError(err, false, totalWait) + } + return nil, CoalesceResult{Waited: waited, WaitDuration: totalWait}, err } c.inflightMu.Lock() @@ -229,25 +277,35 @@ func (c *Cache) Coalesce(ctx context.Context, key string, fetch func(context.Con c.inflightMu.Unlock() if ok { + waited = true + waitStart := time.Now() if err := ctx.Err(); err != nil { - return nil, err + waitDuration := time.Since(waitStart) + totalWait += waitDuration + return nil, CoalesceResult{Coalesced: true, Waited: true, WaitDuration: totalWait}, + attributedCoalescingError(err, true, totalWait) } select { case <-req.done: + totalWait += time.Since(waitStart) if err := ctx.Err(); err != nil { - return nil, err + return nil, CoalesceResult{Coalesced: true, Waited: true, WaitDuration: totalWait}, + attributedCoalescingError(err, true, totalWait) } if req.err != nil { if errors.Is(req.err, context.Canceled) || errors.Is(req.err, context.DeadlineExceeded) { continue } - return nil, req.err + return nil, CoalesceResult{Coalesced: true, Waited: true, WaitDuration: totalWait}, + attributedCoalescingError(req.err, true, totalWait) } data := make([]byte, len(req.result)) copy(data, req.result) - return data, nil + return data, CoalesceResult{Coalesced: true, Waited: true, WaitDuration: totalWait}, nil case <-ctx.Done(): - return nil, ctx.Err() + totalWait += time.Since(waitStart) + return nil, CoalesceResult{Coalesced: true, Waited: true, WaitDuration: totalWait}, + attributedCoalescingError(ctx.Err(), true, totalWait) } } @@ -265,12 +323,15 @@ func (c *Cache) Coalesce(ctx context.Context, key string, fetch func(context.Con close(req.done) if err != nil { - return nil, err + if waited { + err = attributedCoalescingError(err, false, totalWait) + } + return nil, CoalesceResult{Waited: waited, WaitDuration: totalWait}, err } result := make([]byte, len(data)) copy(result, data) - return result, nil + return result, CoalesceResult{Waited: waited, WaitDuration: totalWait}, nil } } diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 453e105..c949234 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -303,6 +303,93 @@ func TestCache_CoalescingConcurrent(t *testing.T) { } } +func TestCache_CoalesceWithStatusAttributesFollowers(t *testing.T) { + tests := []struct { + name string + leaderErr error + }{ + {name: "success"}, + {name: "error", leaderErr: errors.New("upstream failed")}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := New(time.Second, false) + defer c.Close() + started := make(chan struct{}) + release := make(chan struct{}) + leaderDone := make(chan error, 1) + go func() { + _, status, err := c.CoalesceWithStatus(context.Background(), "key", func(context.Context) ([]byte, error) { + close(started) + <-release + return []byte("shared"), tt.leaderErr + }) + if status.Coalesced { + leaderDone <- errors.New("leader reported as coalesced") + return + } + leaderDone <- err + }() + <-started + + type outcome struct { + data []byte + status CoalesceResult + err error + } + followerDone := make(chan outcome, 1) + go func() { + data, status, err := c.CoalesceWithStatus(context.Background(), "key", func(context.Context) ([]byte, error) { + return nil, errors.New("follower fetch ran") + }) + followerDone <- outcome{data: data, status: status, err: err} + }() + + deadline := time.Now().Add(time.Second) + for { + c.inflightMu.Lock() + req := c.inflight["key"] + joined := req != nil && req.followers == 1 + c.inflightMu.Unlock() + if joined { + break + } + if time.Now().After(deadline) { + t.Fatal("follower did not join leader") + } + runtime.Gosched() + } + close(release) + + if err := <-leaderDone; !errors.Is(err, tt.leaderErr) { + t.Fatalf("leader error = %v, want %v", err, tt.leaderErr) + } + follower := <-followerDone + if !follower.status.Coalesced || follower.status.WaitDuration < 0 { + t.Fatalf("missing follower attribution: %+v", follower.status) + } + if tt.leaderErr == nil { + if follower.err != nil || string(follower.data) != "shared" { + t.Fatalf("unexpected follower success data=%q err=%v", follower.data, follower.err) + } + return + } + if !errors.Is(follower.err, tt.leaderErr) { + t.Fatalf("coalesced error lost cause: %v", follower.err) + } + var marker interface { + Coalesced() bool + CoalescedWaitDuration() time.Duration + } + if !errors.As(follower.err, &marker) || + !marker.Coalesced() || + marker.CoalescedWaitDuration() != follower.status.WaitDuration { + t.Fatalf("coalesced error lost attribution: status=%+v err=%v", follower.status, follower.err) + } + }) + } +} + func TestCache_ContextCancellation(t *testing.T) { c := New(time.Second, false) defer c.Close() @@ -376,10 +463,11 @@ func TestCache_LiveFollowerRetriesAfterLeaderDeadline(t *testing.T) { followerDone := make(chan struct{}) var followerData []byte + var followerStatus CoalesceResult var followerErr error var followerFetches atomic.Int32 go func() { - followerData, followerErr = c.Coalesce(context.Background(), "key1", func(context.Context) ([]byte, error) { + followerData, followerStatus, followerErr = c.CoalesceWithStatus(context.Background(), "key1", func(context.Context) ([]byte, error) { followerFetches.Add(1) return []byte("fresh"), nil }) @@ -416,6 +504,76 @@ func TestCache_LiveFollowerRetriesAfterLeaderDeadline(t *testing.T) { if followerFetches.Load() != 1 { t.Fatalf("follower fetch ran %d times", followerFetches.Load()) } + if !followerStatus.Waited || followerStatus.Coalesced || followerStatus.WaitDuration <= 0 { + t.Fatalf("replacement leader lost prior follower attribution: %+v", followerStatus) + } +} + +func TestCache_ReplacementLeaderErrorPreservesWaitAttribution(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 nil, ctx.Err() + }) + leaderDone <- err + }() + <-leaderStarted + + replacementErr := errors.New("replacement failed") + type outcome struct { + status CoalesceResult + err error + } + followerDone := make(chan outcome, 1) + go func() { + _, status, err := c.CoalesceWithStatus(context.Background(), "key1", func(context.Context) ([]byte, error) { + return nil, replacementErr + }) + followerDone <- outcome{status: status, err: err} + }() + + 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("follower did not join leader") + } + runtime.Gosched() + } + + cancelLeader() + if err := <-leaderDone; !errors.Is(err, context.Canceled) { + t.Fatalf("leader error = %v, want cancellation", err) + } + follower := <-followerDone + if !errors.Is(follower.err, replacementErr) || + !follower.status.Waited || + follower.status.Coalesced || + follower.status.WaitDuration <= 0 { + t.Fatalf("replacement result lost wait attribution: status=%+v err=%v", follower.status, follower.err) + } + var marker interface { + Coalesced() bool + CoalescedWaitDuration() time.Duration + } + if !errors.As(follower.err, &marker) || + marker.Coalesced() || + marker.CoalescedWaitDuration() != follower.status.WaitDuration { + t.Fatalf("replacement error lost wait attribution: status=%+v err=%v", follower.status, follower.err) + } } func TestCache_DataIsolation(t *testing.T) { diff --git a/internal/health/health.go b/internal/health/health.go index a32763a..d680476 100644 --- a/internal/health/health.go +++ b/internal/health/health.go @@ -18,8 +18,9 @@ type Checker interface { // Response is the JSON body returned by the health endpoint. type Response struct { - Status string `json:"status"` - Error string `json:"error,omitempty"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + Details map[string]any `json:"details,omitempty"` } // Server is a lightweight HTTP server that exposes a /health endpoint. @@ -50,9 +51,10 @@ func (s *Server) handleHealth(checker Checker) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - if err := checker.Healthy(); err != nil { + status, details, healthErr := healthReport(checker) + if healthErr != nil { w.WriteHeader(http.StatusServiceUnavailable) - resp := Response{Status: "unhealthy", Error: err.Error()} + resp := Response{Status: "unhealthy", Error: healthErr.Error(), Details: details} if encErr := json.NewEncoder(w).Encode(resp); encErr != nil { s.logger.Error("failed to encode health response", "error", encErr) } @@ -60,13 +62,30 @@ func (s *Server) handleHealth(checker Checker) http.HandlerFunc { } w.WriteHeader(http.StatusOK) - resp := Response{Status: "ok"} + resp := Response{Status: status, Details: details} if err := json.NewEncoder(w).Encode(resp); err != nil { s.logger.Error("failed to encode health response", "error", err) } } } +func healthReport(checker Checker) (string, map[string]any, error) { + if reporter, ok := checker.(interface { + HealthReport() (string, map[string]any, error) + }); ok { + return reporter.HealthReport() + } + + status := "ok" + var details map[string]any + if reporter, ok := checker.(interface { + HealthStatus() (string, map[string]any) + }); ok { + status, details = reporter.HealthStatus() + } + return status, details, checker.Healthy() +} + // ListenAndServe starts the health server. It blocks until the server // is shut down or encounters a fatal error. Use Listen + Serve to // separate binding from serving, which allows detecting bind errors early. diff --git a/internal/health/health_test.go b/internal/health/health_test.go index 85d209c..542ae22 100644 --- a/internal/health/health_test.go +++ b/internal/health/health_test.go @@ -17,10 +17,41 @@ type mockChecker struct { err error } +type diagnosticChecker struct { + mockChecker + status string + details map[string]any +} + +type atomicDiagnosticChecker struct { + healthyCalls int + statusCalls int + reportCalls int +} + func (m *mockChecker) Healthy() error { return m.err } +func (c *diagnosticChecker) HealthStatus() (string, map[string]any) { + return c.status, c.details +} + +func (c *atomicDiagnosticChecker) Healthy() error { + c.healthyCalls++ + return fmt.Errorf("separate health snapshot used") +} + +func (c *atomicDiagnosticChecker) HealthStatus() (string, map[string]any) { + c.statusCalls++ + return "unhealthy", nil +} + +func (c *atomicDiagnosticChecker) HealthReport() (string, map[string]any, error) { + c.reportCalls++ + return "degraded", map[string]any{"total_retries": uint64(2)}, nil +} + func TestHandleHealth_OK(t *testing.T) { checker := &mockChecker{err: nil} srv := NewServer(":0", checker, slog.Default()) @@ -46,6 +77,85 @@ func TestHandleHealth_OK(t *testing.T) { } } +func TestHandleHealth_DegradedDiagnosticsRemainAvailable(t *testing.T) { + checker := &diagnosticChecker{ + status: "degraded", + details: map[string]any{ + "total_retries": uint64(2), + "recovered_retries": uint64(2), + "sustained_degradation": true, + }, + } + srv := NewServer(":0", checker, slog.Default()) + req := httptest.NewRequest(http.MethodGet, "/health", nil) + w := httptest.NewRecorder() + + srv.httpServer.Handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("degraded health returned status %d", w.Code) + } + var resp Response + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if resp.Status != "degraded" || + resp.Details["total_retries"] != float64(2) || + resp.Details["sustained_degradation"] != true { + t.Fatalf("unexpected degraded response: %+v", resp) + } +} + +func TestHandleHealth_PrefersAtomicReport(t *testing.T) { + checker := &atomicDiagnosticChecker{} + srv := NewServer(":0", checker, slog.Default()) + req := httptest.NewRequest(http.MethodGet, "/health", nil) + w := httptest.NewRecorder() + + srv.httpServer.Handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("atomic health report returned status %d", w.Code) + } + var resp Response + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if resp.Status != "degraded" || resp.Details["total_retries"] != float64(2) { + t.Fatalf("unexpected atomic report: %+v", resp) + } + if checker.reportCalls != 1 || checker.healthyCalls != 0 || checker.statusCalls != 0 { + t.Fatalf("health endpoint mixed snapshots: %+v", checker) + } +} + +func TestHandleHealth_UnhealthyIncludesDiagnostics(t *testing.T) { + checker := &diagnosticChecker{ + mockChecker: mockChecker{err: fmt.Errorf("upstream unavailable")}, + status: "degraded", + details: map[string]any{ + "consecutive_final_failures": float64(3), + }, + } + srv := NewServer(":0", checker, slog.Default()) + req := httptest.NewRequest(http.MethodGet, "/health", nil) + w := httptest.NewRecorder() + + srv.httpServer.Handler.ServeHTTP(w, req) + + if w.Code != http.StatusServiceUnavailable { + t.Fatalf("unhealthy status = %d", w.Code) + } + var resp Response + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if resp.Status != "unhealthy" || + resp.Details["consecutive_final_failures"] != float64(3) { + t.Fatalf("unhealthy response omitted diagnostics: %+v", resp) + } +} + func TestHandleHealth_Unhealthy(t *testing.T) { checker := &mockChecker{err: fmt.Errorf("upstream unreachable")} srv := NewServer(":0", checker, slog.Default()) diff --git a/internal/modbus/client.go b/internal/modbus/client.go index 0265459..9055e2d 100644 --- a/internal/modbus/client.go +++ b/internal/modbus/client.go @@ -33,6 +33,8 @@ type clientSession interface { type sessionFactory func() (clientSession, requestClient) +const degradedHealthWindow = time.Minute + type guardedConn struct { net.Conn @@ -216,6 +218,18 @@ func (c *tcpSession) BeginRequest(ctx context.Context, attemptTimeout time.Durat } } +// HealthStats is a snapshot of upstream reliability diagnostics. +type HealthStats struct { + LastFirstAttemptSuccess time.Time + LastSuccessfulRequest time.Time + ConsecutiveFirstAttemptFailure int + ConsecutiveFinalFailure int + TotalRetries uint64 + RecoveredRetries uint64 + Degraded bool + SustainedDegradation bool +} + // Client wraps a Modbus TCP client with classified retry behavior. type Client struct { address string @@ -223,6 +237,9 @@ type Client struct { requestDelay time.Duration connectDelay time.Duration logger *slog.Logger + now func() time.Time + degradedWindow time.Duration + beforeAcquire func() owner chan struct{} session clientSession @@ -230,9 +247,16 @@ type Client struct { newSession sessionFactory nextRequestAt time.Time - healthMu sync.RWMutex - lastErr error - connected bool + healthMu sync.RWMutex + lastErr error + connected bool + firstFailureAt time.Time + lastFirstSuccess time.Time + lastSuccess time.Time + firstFailures int + finalFailures int + totalRetries uint64 + recoveredRetries uint64 } // NewClient creates a new Modbus TCP client. @@ -243,6 +267,8 @@ func NewClient(address string, attemptTimeout, requestDelay, connectDelay time.D requestDelay: requestDelay, connectDelay: connectDelay, logger: logger, + now: time.Now, + degradedWindow: degradedHealthWindow, owner: make(chan struct{}, 1), } c.owner <- struct{}{} @@ -349,10 +375,14 @@ func (c *Client) release() { c.owner <- struct{}{} } -// Healthy reports the last observed upstream state. +// Healthy reports final upstream failures while recovered retries remain available. func (c *Client) Healthy() error { c.healthMu.RLock() defer c.healthMu.RUnlock() + return c.healthyLocked() +} + +func (c *Client) healthyLocked() error { if c.lastErr != nil { return c.lastErr } @@ -362,45 +392,134 @@ func (c *Client) Healthy() error { return nil } +// HealthStats returns retry, failure streak, and recovery state. +func (c *Client) HealthStats() HealthStats { + c.healthMu.RLock() + defer c.healthMu.RUnlock() + return c.healthStatsLocked(c.now()) +} + +func (c *Client) healthStatsLocked(now time.Time) HealthStats { + degraded := !c.firstFailureAt.IsZero() + sustained := degraded && + (c.degradedWindow <= 0 || now.Sub(c.firstFailureAt) >= c.degradedWindow) + return HealthStats{ + LastFirstAttemptSuccess: c.lastFirstSuccess, + LastSuccessfulRequest: c.lastSuccess, + ConsecutiveFirstAttemptFailure: c.firstFailures, + ConsecutiveFinalFailure: c.finalFailures, + TotalRetries: c.totalRetries, + RecoveredRetries: c.recoveredRetries, + Degraded: degraded, + SustainedDegradation: sustained, + } +} + +// HealthStatus supplies reliability details to the HTTP health endpoint. +func (c *Client) HealthStatus() (string, map[string]any) { + c.healthMu.RLock() + defer c.healthMu.RUnlock() + return healthStatus(c.healthStatsLocked(c.now())) +} + +// HealthReport returns status, diagnostics, and availability from one snapshot. +func (c *Client) HealthReport() (string, map[string]any, error) { + c.healthMu.RLock() + defer c.healthMu.RUnlock() + status, details := healthStatus(c.healthStatsLocked(c.now())) + return status, details, c.healthyLocked() +} + +func healthStatus(stats HealthStats) (string, map[string]any) { + status := "ok" + if stats.SustainedDegradation { + status = "degraded" + } + return status, map[string]any{ + "last_first_attempt_success": healthTimestamp(stats.LastFirstAttemptSuccess), + "last_successful_request": healthTimestamp(stats.LastSuccessfulRequest), + "consecutive_first_attempt_failures": stats.ConsecutiveFirstAttemptFailure, + "consecutive_final_failures": stats.ConsecutiveFinalFailure, + "total_retries": stats.TotalRetries, + "recovered_retries": stats.RecoveredRetries, + "degraded": stats.Degraded, + "sustained_degradation": stats.SustainedDegradation, + } +} + +func healthTimestamp(timestamp time.Time) any { + if timestamp.IsZero() { + return nil + } + return timestamp +} + // 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) { + start := c.now() + timing := requestTiming{} if err := ValidateRequest(req); err != nil { - return nil, requestError(err, 0) + timing.total = c.now().Sub(start) + return nil, requestError(err, 0, timing) } + queueStart := c.now() + if c.beforeAcquire != nil { + c.beforeAcquire() + } if err := c.acquire(ctx); err != nil { - return nil, requestError(err, 0) + timing.queue = c.now().Sub(queueStart) + timing.total = c.now().Sub(start) + return nil, requestError(err, 0, timing) } + timing.queue = c.now().Sub(queueStart) defer c.release() var lastErr error requestSlotReady := false for attempt := 1; attempt <= 2; attempt++ { if c.session == nil { + reconnectStart := c.now() err := c.connectLocked(ctx) + timing.reconnect += c.now().Sub(reconnectStart) if err != nil { lastErr = err willRetry := attempt == 1 && IsReadFunction(req.FunctionCode) && isTransportError(err) && contextError(ctx) == nil + timing.total = c.now().Sub(start) + if attempt == 1 && isTransportError(err) { + c.recordFirstFailure() + } if willRetry { - c.logger.Debug("upstream connect failed, retrying read", "error", err) + c.recordRetry() + c.logFailure(req, attempt, timing, err, true) continue } - return nil, c.finalError(lastErr, attempt) + c.logFailure(req, attempt, timing, err, false) + return nil, c.finalError(lastErr, attempt, timing) } } if !requestSlotReady { + queueStart = c.now() if err := c.waitForRequestSlot(ctx); err != nil { - return nil, c.finalError(err, attempt-1) + timing.queue += c.now().Sub(queueStart) + timing.total = c.now().Sub(start) + c.logFailure(req, attempt-1, timing, err, false) + return nil, c.finalError(err, attempt-1, timing) } + timing.queue += c.now().Sub(queueStart) requestSlotReady = true } if err := contextError(ctx); err != nil { - return nil, c.finalError(err, attempt-1) + timing.total = c.now().Sub(start) + c.logFailure(req, attempt-1, timing, err, false) + return nil, c.finalError(err, attempt-1, timing) } c.session.SetSlave(req.SlaveID) if err := contextError(ctx); err != nil { - return nil, c.finalError(err, attempt-1) + timing.total = c.now().Sub(start) + c.logFailure(req, attempt-1, timing, err, false) + return nil, c.finalError(err, attempt-1, timing) } attemptTimeout := c.socketTimeoutFor(ctx) finishRequest := c.session.BeginRequest(ctx, attemptTimeout) @@ -408,12 +527,14 @@ func (c *Client) Execute(ctx context.Context, req *Request) ([]byte, error) { if finishErr := finishRequest(); finishErr != nil { c.logger.Debug("finishing canceled upstream request failed", "error", finishErr) } - return nil, c.finalError(err, attempt-1) + timing.total = c.now().Sub(start) + c.logFailure(req, attempt-1, timing, err, false) + return nil, c.finalError(err, attempt-1, timing) } - attemptStart := time.Now() + attemptStart := c.now() resp, err := c.executeRequest(ctx, req) responseCompletedAt := time.Now() - attemptDuration := time.Since(attemptStart) + timing.attempt += c.now().Sub(attemptStart) if finishErr := finishRequest(); finishErr != nil { c.logger.Debug("finishing upstream request failed", "error", finishErr) } @@ -430,38 +551,44 @@ func (c *Client) Execute(ctx context.Context, req *Request) ([]byte, error) { c.nextRequestAt = responseCompletedAt.Add(c.requestDelay) } if ctxErr := contextError(ctx); ctxErr != nil { - return nil, c.finalError(ctxErr, attempt) + timing.total = c.now().Sub(start) + c.logFailure(req, attempt, timing, ctxErr, false) + return nil, c.finalError(ctxErr, attempt, timing) } - 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, - ) + completedAt := c.now() + timing.total = completedAt.Sub(start) + timing = completeTiming(timing) + c.recordSuccess(attempt, completedAt) + c.logSuccess(req, attempt, timing) return resp, nil } lastErr = err kind, _ = classifyError(err) + timing.total = c.now().Sub(start) willRetry := attempt == 1 && IsReadFunction(req.FunctionCode) && (kind == ErrorTransportTimeout || kind == ErrorTransportClosed) && contextError(ctx) == nil + if attempt == 1 && (kind == ErrorTransportTimeout || kind == ErrorTransportClosed) { + c.recordFirstFailure() + } if willRetry { c.disconnectLocked() - c.logger.Debug("upstream request failed, retrying read", "error", err) + c.recordRetry() + c.logFailure(req, attempt, timing, err, true) continue } if kind != ErrorProtocolException { c.disconnectLocked() } - return nil, c.finalError(lastErr, attempt) + c.logFailure(req, attempt, timing, err, false) + return nil, c.finalError(lastErr, attempt, timing) } - return nil, c.finalError(lastErr, 2) + timing.total = c.now().Sub(start) + return nil, c.finalError(lastErr, 2, timing) } func (c *Client) waitForRequestSlot(ctx context.Context) error { @@ -525,28 +652,127 @@ func contextError(ctx context.Context) error { return nil } -func (c *Client) finalError(err error, attempts int) error { - reqErr := requestError(err, attempts) +func completeTiming(timing requestTiming) requestTiming { + const maxDuration = time.Duration(1<<63 - 1) + var componentTotal time.Duration + for _, duration := range []time.Duration{timing.queue, timing.attempt, timing.reconnect} { + if duration <= 0 { + continue + } + if duration > maxDuration-componentTotal { + componentTotal = maxDuration + break + } + componentTotal += duration + } + if timing.total < componentTotal { + timing.total = componentTotal + } + return timing +} + +func (c *Client) finalError(err error, attempts int, timing requestTiming) error { + reqErr := requestError(err, attempts, completeTiming(timing)) if reqErr.Kind == ErrorProtocolException { c.healthMu.Lock() c.lastErr = nil + c.finalFailures = 0 c.connected = true c.healthMu.Unlock() } else if reqErr.Kind != ErrorContextCanceled { c.healthMu.Lock() c.lastErr = reqErr + if reqErr.Kind == ErrorTransportTimeout || reqErr.Kind == ErrorTransportClosed { + c.finalFailures++ + } c.healthMu.Unlock() } return reqErr } -func (c *Client) recordSuccess() { +func (c *Client) recordFirstFailure() { + c.healthMu.Lock() + if c.firstFailureAt.IsZero() { + c.firstFailureAt = c.now() + } + c.firstFailures++ + c.healthMu.Unlock() +} + +func (c *Client) recordRetry() { c.healthMu.Lock() + c.totalRetries++ + c.healthMu.Unlock() +} + +func (c *Client) recordSuccess(attempts int, now time.Time) { + c.healthMu.Lock() + if attempts == 1 { + c.lastFirstSuccess = now + c.firstFailureAt = time.Time{} + c.firstFailures = 0 + } else { + c.recoveredRetries++ + } + c.lastSuccess = now c.lastErr = nil + c.finalFailures = 0 c.connected = true c.healthMu.Unlock() } +func (c *Client) logFailure(req *Request, attempts int, timing requestTiming, err error, willRetry bool) { + timing = completeTiming(timing) + kind, exceptionCode := classifyError(err) + args := []any{ + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", IsWriteFunction(req.FunctionCode), + "attempt", attempts, + "attempts", attempts, + "queue_duration", timing.queue, + "attempt_duration", timing.attempt, + "reconnect_duration", timing.reconnect, + "total_duration", timing.total, + "error_kind", kind, + "will_retry", willRetry, + } + if !IsWriteFunction(req.FunctionCode) { + args = append(args, "error", err) + } + if exceptionCode != 0 { + args = append(args, "exception_code", fmt.Sprintf("0x%02X", exceptionCode)) + } + if !willRetry { + args = append(args, "downstream_exception", fmt.Sprintf("0x%02X", DownstreamException(err))) + } + if willRetry { + c.logger.Debug("upstream request attempt failed", args...) + return + } + c.logger.Debug("upstream request failed", args...) +} + +func (c *Client) logSuccess(req *Request, attempts int, timing requestTiming) { + c.logger.Debug("upstream request completed", + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", IsWriteFunction(req.FunctionCode), + "attempt", attempts, + "attempts", attempts, + "queue_duration", timing.queue, + "attempt_duration", timing.attempt, + "reconnect_duration", timing.reconnect, + "total_duration", timing.total, + "error_kind", "", + "will_retry", false, + ) +} + func isTransportError(err error) bool { kind, _ := classifyError(err) return kind == ErrorTransportTimeout || kind == ErrorTransportClosed diff --git a/internal/modbus/client_test.go b/internal/modbus/client_test.go index 00e9282..150169a 100644 --- a/internal/modbus/client_test.go +++ b/internal/modbus/client_test.go @@ -1,8 +1,10 @@ package modbus import ( + "bytes" "context" "encoding/binary" + "encoding/json" "errors" "io" "log/slog" @@ -118,6 +120,7 @@ func (c *fakeSession) BeginRequest(ctx context.Context, _ time.Duration) func() type fakeSessionSet struct { client *fakeRequestClient connectErrs []error + connectHook func(context.Context) sessions atomic.Int32 connects atomic.Int32 closes atomic.Int32 @@ -130,7 +133,13 @@ func (s *fakeSessionSet) factory() (clientSession, requestClient) { if index < len(s.connectErrs) { err = s.connectErrs[index] } - return &fakeSession{connectErr: err, finishErr: s.finishErr, connects: &s.connects, closes: &s.closes}, s.client + return &fakeSession{ + connectErr: err, + connectHook: s.connectHook, + finishErr: s.finishErr, + connects: &s.connects, + closes: &s.closes, + }, s.client } func newFakeClient(results []fakeResult) (*Client, *fakeSessionSet) { @@ -155,6 +164,68 @@ func writeRequest() *Request { } } +type testClock struct { + mu sync.Mutex + now time.Time +} + +func (c *testClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *testClock) Advance(duration time.Duration) { + c.mu.Lock() + c.now = c.now.Add(duration) + c.mu.Unlock() +} + +func newDiagnosticLogger() (*bytes.Buffer, *slog.Logger) { + var output bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})) + return &output, logger +} + +func findLogEntry(t *testing.T, output *bytes.Buffer, message string) map[string]any { + t.Helper() + for _, line := range bytes.Split(bytes.TrimSpace(output.Bytes()), []byte{'\n'}) { + var entry map[string]any + if err := json.Unmarshal(line, &entry); err != nil { + t.Fatalf("decode log entry: %v", err) + } + if entry["msg"] == message { + return entry + } + } + t.Fatalf("log message %q not found in %s", message, output.String()) + return nil +} + +func requireLifecycleFields(t *testing.T, entry map[string]any) { + t.Helper() + for _, field := range []string{ + "slave_id", "func", "addr", "qty", "write", "attempt", "attempts", + "queue_duration", "attempt_duration", "reconnect_duration", + "total_duration", "error_kind", "will_retry", + } { + if _, ok := entry[field]; !ok { + t.Errorf("log entry missing %q: %+v", field, entry) + } + } +} + +func requireDurationField(t *testing.T, entry map[string]any, field string, want time.Duration) { + t.Helper() + value, ok := entry[field].(float64) + if !ok { + t.Fatalf("%s has type %T, want JSON number", field, entry[field]) + } + if got := time.Duration(value); got != want { + t.Fatalf("%s = %v, want %v", field, got, want) + } +} + func TestClient_ProtocolExceptionIsNotRetried(t *testing.T) { client, sessions := newFakeClient([]fakeResult{{err: &gridmodbus.Error{ FunctionCode: FuncReadHoldingRegisters | 0x80, @@ -256,6 +327,221 @@ func TestClient_WriteTimeoutIsNotRetried(t *testing.T) { } } +func TestClient_RetryDiagnosticsSeparateTiming(t *testing.T) { + clock := &testClock{ + now: time.Unix(1000, 0), + } + queueStarted := make(chan struct{}) + output, logger := newDiagnosticLogger() + client, sessions := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}, complete: func() { clock.Advance(3 * time.Second) }}, + {data: []byte{0x12, 0x34}, complete: func() { clock.Advance(3 * time.Second) }}, + }) + client.logger = logger + client.now = clock.Now + client.beforeAcquire = func() { close(queueStarted) } + sessions.connectHook = func(context.Context) { clock.Advance(2 * time.Second) } + + <-client.owner + result := make(chan error, 1) + go func() { + _, err := client.Execute(context.Background(), readRequest()) + result <- err + }() + <-queueStarted + clock.Advance(time.Second) + client.release() + if err := <-result; err != nil { + t.Fatalf("execute: %v", err) + } + + firstFailure := findLogEntry(t, output, "upstream request attempt failed") + requireLifecycleFields(t, firstFailure) + if firstFailure["level"] != "DEBUG" || + firstFailure["attempts"] != float64(1) || + firstFailure["error_kind"] != string(ErrorTransportTimeout) || + firstFailure["will_retry"] != true { + t.Fatalf("unexpected first failure log: %+v", firstFailure) + } + requireDurationField(t, firstFailure, "queue_duration", time.Second) + requireDurationField(t, firstFailure, "attempt_duration", 3*time.Second) + requireDurationField(t, firstFailure, "reconnect_duration", 2*time.Second) + requireDurationField(t, firstFailure, "total_duration", 6*time.Second) + + recovered := findLogEntry(t, output, "upstream request completed") + requireLifecycleFields(t, recovered) + if recovered["level"] != "DEBUG" || + recovered["attempts"] != float64(2) || + recovered["error_kind"] != "" || + recovered["will_retry"] != false { + t.Fatalf("unexpected recovered retry log: %+v", recovered) + } + requireDurationField(t, recovered, "queue_duration", time.Second) + requireDurationField(t, recovered, "attempt_duration", 6*time.Second) + requireDurationField(t, recovered, "reconnect_duration", 4*time.Second) + requireDurationField(t, recovered, "total_duration", 11*time.Second) +} + +func TestClient_FinalFailureLogHasCompleteClassification(t *testing.T) { + output, logger := newDiagnosticLogger() + client, _ := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {err: fakeTimeoutError{}}, + }) + client.logger = logger + + _, err := client.Execute(t.Context(), readRequest()) + if err == nil { + t.Fatal("expected final failure") + } + entry := findLogEntry(t, output, "upstream request failed") + requireLifecycleFields(t, entry) + if entry["level"] != "DEBUG" || + entry["attempts"] != float64(2) || + entry["error_kind"] != string(ErrorTransportTimeout) || + entry["will_retry"] != false || + entry["downstream_exception"] != "0x0B" { + t.Fatalf("unexpected final failure log: %+v", entry) + } +} + +func TestClient_ProtocolExceptionLogPreservesCode(t *testing.T) { + output, logger := newDiagnosticLogger() + client, sessions := newFakeClient([]fakeResult{{err: &gridmodbus.Error{ + FunctionCode: FuncReadHoldingRegisters | 0x80, + ExceptionCode: ExcIllegalAddress, + }}}) + client.logger = logger + + _, err := client.Execute(t.Context(), readRequest()) + if ErrorKindOf(err) != ErrorProtocolException { + t.Fatalf("expected protocol exception, got %v", err) + } + entry := findLogEntry(t, output, "upstream request failed") + requireLifecycleFields(t, entry) + if entry["level"] != "DEBUG" || + entry["attempts"] != float64(1) || + entry["exception_code"] != "0x02" || + entry["downstream_exception"] != "0x02" || + entry["will_retry"] != false { + t.Fatalf("unexpected protocol exception log: %+v", entry) + } + if sessions.client.callCount() != 1 || sessions.sessions.Load() != 1 { + t.Fatalf("protocol exception reconnected or retried: calls=%d sessions=%d", sessions.client.callCount(), sessions.sessions.Load()) + } +} + +func TestClient_WriteDiagnosticsNeverLogPayload(t *testing.T) { + const payload = "sensitivepayload1234" + output, logger := newDiagnosticLogger() + client, _ := newFakeClient([]fakeResult{{err: errors.New(payload)}}) + client.logger = logger + req := &Request{ + SlaveID: 7, + FunctionCode: FuncWriteMultipleRegs, + Address: 42, + Quantity: 10, + Data: []byte(payload), + } + + if _, err := client.Execute(t.Context(), req); err == nil { + t.Fatal("expected write failure") + } + if bytes.Contains(output.Bytes(), []byte(payload)) { + t.Fatalf("write payload leaked into logs: %s", output.String()) + } + entry := findLogEntry(t, output, "upstream request failed") + requireLifecycleFields(t, entry) + if entry["write"] != true || entry["slave_id"] != float64(7) || entry["addr"] != float64(42) { + t.Fatalf("write identity missing from diagnostics: %+v", entry) + } +} + +func TestClient_HealthDiagnosticsRetainAndClearRecoveredRetry(t *testing.T) { + clock := &testClock{now: time.Unix(1000, 0)} + client, _ := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {data: []byte{0x12, 0x34}}, + {data: []byte{0x56, 0x78}}, + }) + client.now = clock.Now + + if _, err := client.Execute(t.Context(), readRequest()); err != nil { + t.Fatalf("recovered retry: %v", err) + } + stats := client.HealthStats() + if stats.TotalRetries != 1 || + stats.RecoveredRetries != 1 || + stats.ConsecutiveFirstAttemptFailure != 1 || + stats.ConsecutiveFinalFailure != 0 || + !stats.Degraded || + stats.SustainedDegradation { + t.Fatalf("unexpected recovered retry diagnostics: %+v", stats) + } + if err := client.Healthy(); err != nil { + t.Fatalf("isolated recovered retry failed health: %v", err) + } + status, details := client.HealthStatus() + if status != "ok" || details["degraded"] != true || details["sustained_degradation"] != false { + t.Fatalf("isolated retry status=%q details=%+v", status, details) + } + + clock.Advance(degradedHealthWindow) + status, details = client.HealthStatus() + if status != "degraded" || details["sustained_degradation"] != true { + t.Fatalf("sustained degradation status=%q details=%+v", status, details) + } + if err := client.Healthy(); err != nil { + t.Fatalf("sustained recovered degradation should remain available: %v", err) + } + + clock.Advance(time.Second) + if _, err := client.Execute(t.Context(), readRequest()); err != nil { + t.Fatalf("first-attempt success: %v", err) + } + stats = client.HealthStats() + if stats.Degraded || + stats.SustainedDegradation || + stats.ConsecutiveFirstAttemptFailure != 0 || + stats.TotalRetries != 1 || + stats.RecoveredRetries != 1 || + stats.LastFirstAttemptSuccess.IsZero() || + stats.LastSuccessfulRequest.IsZero() { + t.Fatalf("first-attempt success did not clear degradation: %+v", stats) + } +} + +func TestClient_DiagnosticCountersTrackConsecutiveFinalFailures(t *testing.T) { + client, _ := newFakeClient([]fakeResult{ + {err: fakeTimeoutError{}}, + {err: fakeTimeoutError{}}, + {err: fakeTimeoutError{}}, + {err: fakeTimeoutError{}}, + {data: []byte{0x12, 0x34}}, + }) + + for range 2 { + if _, err := client.Execute(t.Context(), readRequest()); err == nil { + t.Fatal("expected final failure") + } + } + stats := client.HealthStats() + if stats.TotalRetries != 2 || + stats.RecoveredRetries != 0 || + stats.ConsecutiveFirstAttemptFailure != 2 || + stats.ConsecutiveFinalFailure != 2 { + t.Fatalf("unexpected failure counters: %+v", stats) + } + + if _, err := client.Execute(t.Context(), readRequest()); err != nil { + t.Fatalf("first-attempt recovery: %v", err) + } + stats = client.HealthStats() + if stats.ConsecutiveFirstAttemptFailure != 0 || stats.ConsecutiveFinalFailure != 0 { + t.Fatalf("success did not clear consecutive counters: %+v", stats) + } +} + func TestClient_ExpiredQueueRequestNeverExecutes(t *testing.T) { started := make(chan struct{}) release := make(chan struct{}) @@ -304,6 +590,9 @@ func TestClient_ContextDeadlineCapsAttempt(t *testing.T) { if ErrorKindOf(err) != ErrorContextDeadline { t.Fatalf("expected context deadline classification, got %s", ErrorKindOf(err)) } + if stats := client.HealthStats(); stats.ConsecutiveFinalFailure != 0 { + t.Fatalf("request deadline counted as upstream final failure: %+v", stats) + } if client.session != nil { t.Fatal("deadline-failed connection was retained") } @@ -369,6 +658,9 @@ func TestClient_SuccessRacingCancellationReturnsCancellationAndKeepsConnection(t if client.session == nil { t.Fatal("matching successful response unnecessarily closed connection") } + if stats := client.HealthStats(); stats.ConsecutiveFinalFailure != 0 { + t.Fatalf("post-response cancellation counted as upstream final failure: %+v", stats) + } } func TestClient_SuccessReturnsBeforePacingDelay(t *testing.T) { @@ -417,6 +709,9 @@ func TestClient_NextRequestPacingUsesNextRequestBudget(t *testing.T) { if sessions.client.callCount() != 1 { t.Fatalf("expired paced request made %d wire calls", sessions.client.callCount()) } + if stats := client.HealthStats(); stats.ConsecutiveFinalFailure != 0 { + t.Fatalf("pacing deadline counted as upstream final failure: %+v", stats) + } laterDone := make(chan error, 1) go func() { diff --git a/internal/modbus/errors.go b/internal/modbus/errors.go index fce33bc..05dd315 100644 --- a/internal/modbus/errors.go +++ b/internal/modbus/errors.go @@ -8,6 +8,7 @@ import ( "net" "os" "syscall" + "time" gridmodbus "github.com/grid-x/modbus" ) @@ -24,12 +25,16 @@ const ( ErrorLocal ErrorKind = "local_error" ) -// RequestError carries the classified request failure. +// RequestError carries the classified request failure and its timing. type RequestError struct { - Kind ErrorKind - ExceptionCode byte - Attempts int - Err error + Kind ErrorKind + ExceptionCode byte + Attempts int + QueueDuration time.Duration + AttemptDuration time.Duration + ReconnectDuration time.Duration + TotalDuration time.Duration + Err error } type validationError struct { @@ -162,13 +167,17 @@ func DownstreamException(err error) byte { } } -func requestError(err error, attempts int) *RequestError { +func requestError(err error, attempts int, timing requestTiming) *RequestError { kind, exceptionCode := classifyError(err) return &RequestError{ - Kind: kind, - ExceptionCode: exceptionCode, - Attempts: attempts, - Err: err, + Kind: kind, + ExceptionCode: exceptionCode, + Attempts: attempts, + QueueDuration: timing.queue, + AttemptDuration: timing.attempt, + ReconnectDuration: timing.reconnect, + TotalDuration: timing.total, + Err: err, } } @@ -178,3 +187,10 @@ func newValidationError(exceptionCode byte, format string, args ...any) error { err: fmt.Errorf(format, args...), } } + +type requestTiming struct { + queue time.Duration + attempt time.Duration + reconnect time.Duration + total time.Duration +} diff --git a/internal/modbus/server.go b/internal/modbus/server.go index 70ad507..6a28863 100644 --- a/internal/modbus/server.go +++ b/internal/modbus/server.go @@ -173,6 +173,7 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) { return } + requestStart := time.Now() requestCtx := ctx cancel := func() {} if s.requestTimeout > 0 { @@ -185,11 +186,7 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) { cancel() if err != nil { exceptionCode := DownstreamException(err) - s.logger.Debug("handler error", - "error", err, - "func", fmt.Sprintf("0x%02X", req.FunctionCode), - "exception", fmt.Sprintf("0x%02X", exceptionCode), - ) + s.logHandlerError(req, err, exceptionCode, time.Since(requestStart)) 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) @@ -205,6 +202,63 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) { } } +func (s *Server) logHandlerError(req *Request, err error, downstreamException byte, totalDuration time.Duration) { + var coalescedMarker interface { + Coalesced() bool + CoalescedWaitDuration() time.Duration + } + coalescedWaited := errors.As(err, &coalescedMarker) + coalesced := coalescedWaited && coalescedMarker.Coalesced() + coalescedWait := time.Duration(0) + if coalescedWaited { + coalescedWait = coalescedMarker.CoalescedWaitDuration() + } + + attempts := 0 + queueDuration := coalescedWait + attemptDuration := time.Duration(0) + reconnectDuration := time.Duration(0) + var reqErr *RequestError + hasRequestError := errors.As(err, &reqErr) + if hasRequestError && !coalesced { + attempts = reqErr.Attempts + queueDuration += reqErr.QueueDuration + attemptDuration = reqErr.AttemptDuration + reconnectDuration = reqErr.ReconnectDuration + } + + args := []any{ + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", IsWriteFunction(req.FunctionCode), + "attempt", attempts, + "attempts", attempts, + "queue_duration", queueDuration, + "attempt_duration", attemptDuration, + "reconnect_duration", reconnectDuration, + "total_duration", totalDuration, + "error_kind", ErrorKindOf(err), + "will_retry", false, + "downstream_exception", fmt.Sprintf("0x%02X", downstreamException), + "coalesced", coalesced, + "coalesced_waited", coalescedWaited, + "coalesced_wait_duration", coalescedWait, + } + if !IsWriteFunction(req.FunctionCode) { + args = append(args, "error", err) + } + if reqErr != nil && reqErr.ExceptionCode != 0 { + args = append(args, "exception_code", fmt.Sprintf("0x%02X", reqErr.ExceptionCode)) + } + if ErrorKindOf(err) == ErrorContextCanceled { + s.logger.Debug("request canceled", args...) + return + } + s.logger.Warn("request failed", args...) +} + func (s *Server) readRequest(conn net.Conn) (*Request, error) { // Read MBAP header header := make([]byte, mbapHeaderSize) diff --git a/internal/modbus/server_test.go b/internal/modbus/server_test.go index 9f28e2e..16b179f 100644 --- a/internal/modbus/server_test.go +++ b/internal/modbus/server_test.go @@ -1,8 +1,10 @@ package modbus import ( + "bytes" "context" "encoding/binary" + "errors" "io" "log/slog" "net" @@ -18,6 +20,28 @@ type mockHandler struct { err error } +type testCoalescedError struct { + err error + coalesced bool + wait time.Duration +} + +func (e *testCoalescedError) Error() string { + return e.err.Error() +} + +func (e *testCoalescedError) Unwrap() error { + return e.err +} + +func (e *testCoalescedError) Coalesced() bool { + return e.coalesced +} + +func (e *testCoalescedError) CoalescedWaitDuration() time.Duration { + return e.wait +} + func (h *mockHandler) HandleRequest(ctx context.Context, req *Request) ([]byte, error) { return h.response, h.err } @@ -175,6 +199,200 @@ func TestServer_PreservesUpstreamException(t *testing.T) { } } +func TestServer_CoalescedFailureDoesNotClaimLeaderAttempts(t *testing.T) { + output, logger := newDiagnosticLogger() + server := NewServer(nil, time.Second, logger) + req := &Request{ + SlaveID: 9, + FunctionCode: FuncReadHoldingRegisters, + Address: 400, + Quantity: 2, + } + err := &testCoalescedError{ + err: &RequestError{ + Kind: ErrorTransportTimeout, + Attempts: 2, + QueueDuration: time.Second, + AttemptDuration: 2 * time.Second, + ReconnectDuration: 3 * time.Second, + TotalDuration: 6 * time.Second, + Err: fakeTimeoutError{}, + }, + coalesced: true, + wait: 4 * time.Second, + } + + server.logHandlerError(req, err, ExcGatewayTargetFailed, 5*time.Second) + + entry := findLogEntry(t, output, "request failed") + requireLifecycleFields(t, entry) + if entry["level"] != "WARN" || + entry["attempts"] != float64(0) || + entry["attempt_duration"] != float64(0) || + entry["reconnect_duration"] != float64(0) || + entry["coalesced"] != true || + entry["coalesced_waited"] != true || + entry["downstream_exception"] != "0x0B" { + t.Fatalf("coalesced follower claimed leader diagnostics: %+v", entry) + } + requireDurationField(t, entry, "queue_duration", 4*time.Second) + requireDurationField(t, entry, "coalesced_wait_duration", 4*time.Second) + requireDurationField(t, entry, "total_duration", 5*time.Second) +} + +func TestServer_ReplacementLeaderFailureIncludesWaitAndAttempts(t *testing.T) { + output, logger := newDiagnosticLogger() + server := NewServer(nil, time.Second, logger) + req := &Request{ + SlaveID: 9, + FunctionCode: FuncReadHoldingRegisters, + Address: 400, + Quantity: 2, + } + err := &testCoalescedError{ + err: &RequestError{ + Kind: ErrorTransportTimeout, + Attempts: 2, + QueueDuration: time.Second, + AttemptDuration: 2 * time.Second, + ReconnectDuration: 3 * time.Second, + TotalDuration: 6 * time.Second, + Err: fakeTimeoutError{}, + }, + wait: 4 * time.Second, + } + + server.logHandlerError(req, err, ExcGatewayTargetFailed, 10*time.Second) + + entry := findLogEntry(t, output, "request failed") + if entry["level"] != "WARN" || + entry["attempts"] != float64(2) || + entry["coalesced"] != false || + entry["coalesced_waited"] != true { + t.Fatalf("replacement leader attribution missing: %+v", entry) + } + requireDurationField(t, entry, "queue_duration", 5*time.Second) + requireDurationField(t, entry, "attempt_duration", 2*time.Second) + requireDurationField(t, entry, "reconnect_duration", 3*time.Second) + requireDurationField(t, entry, "total_duration", 10*time.Second) +} + +func TestServer_FinalFailureWarnHasCompleteDiagnostics(t *testing.T) { + output, logger := newDiagnosticLogger() + server := NewServer(nil, time.Second, logger) + req := &Request{ + SlaveID: 9, + FunctionCode: FuncReadHoldingRegisters, + Address: 400, + Quantity: 2, + } + err := &RequestError{ + Kind: ErrorTransportTimeout, + Attempts: 2, + QueueDuration: time.Second, + AttemptDuration: 2 * time.Second, + ReconnectDuration: 3 * time.Second, + TotalDuration: 6 * time.Second, + Err: fakeTimeoutError{}, + } + + server.logHandlerError(req, err, ExcGatewayTargetFailed, 7*time.Second) + + entry := findLogEntry(t, output, "request failed") + requireLifecycleFields(t, entry) + if entry["level"] != "WARN" || + entry["attempts"] != float64(2) || + entry["error_kind"] != string(ErrorTransportTimeout) || + entry["downstream_exception"] != "0x0B" || + entry["will_retry"] != false { + t.Fatalf("unexpected final failure log: %+v", entry) + } + requireDurationField(t, entry, "queue_duration", time.Second) + requireDurationField(t, entry, "attempt_duration", 2*time.Second) + requireDurationField(t, entry, "reconnect_duration", 3*time.Second) + requireDurationField(t, entry, "total_duration", 7*time.Second) +} + +func TestServer_ProtocolExceptionWarnPreservesCode(t *testing.T) { + output, logger := newDiagnosticLogger() + server := NewServer(nil, time.Second, logger) + req := &Request{ + SlaveID: 9, + FunctionCode: FuncReadHoldingRegisters, + Address: 400, + Quantity: 2, + } + err := &RequestError{ + Kind: ErrorProtocolException, + ExceptionCode: ExcIllegalAddress, + Attempts: 1, + Err: &gridmodbus.Error{ + FunctionCode: FuncReadHoldingRegisters | 0x80, + ExceptionCode: ExcIllegalAddress, + }, + } + + server.logHandlerError(req, err, ExcIllegalAddress, time.Second) + + entry := findLogEntry(t, output, "request failed") + requireLifecycleFields(t, entry) + if entry["level"] != "WARN" || + entry["attempts"] != float64(1) || + entry["error_kind"] != string(ErrorProtocolException) || + entry["exception_code"] != "0x02" || + entry["downstream_exception"] != "0x02" { + t.Fatalf("unexpected protocol exception log: %+v", entry) + } +} + +func TestServer_ContextCancellationIsDebug(t *testing.T) { + output, logger := newDiagnosticLogger() + server := NewServer(nil, time.Second, logger) + req := &Request{ + SlaveID: 9, + FunctionCode: FuncReadHoldingRegisters, + Address: 400, + Quantity: 2, + } + + server.logHandlerError(req, context.Canceled, ExcGatewayTargetFailed, time.Second) + + entry := findLogEntry(t, output, "request canceled") + requireLifecycleFields(t, entry) + if entry["level"] != "DEBUG" || entry["error_kind"] != string(ErrorContextCanceled) { + t.Fatalf("unexpected cancellation log: %+v", entry) + } +} + +func TestServer_WriteFailureDoesNotLogRawError(t *testing.T) { + const payload = "sensitivepayload1234" + output, logger := newDiagnosticLogger() + server := NewServer(nil, time.Second, logger) + req := &Request{ + SlaveID: 9, + FunctionCode: FuncWriteMultipleRegs, + Address: 400, + Quantity: 10, + Data: []byte(payload), + } + err := &RequestError{ + Kind: ErrorTransportClosed, + Attempts: 1, + Err: errors.New(payload), + } + + server.logHandlerError(req, err, ExcGatewayTargetFailed, time.Second) + + if bytes.Contains(output.Bytes(), []byte(payload)) { + t.Fatalf("write payload leaked through raw error: %s", output.String()) + } + entry := findLogEntry(t, output, "request failed") + requireLifecycleFields(t, entry) + if _, ok := entry["error"]; ok { + t.Fatalf("write failure retained raw error: %+v", entry) + } +} + func TestServer_MapsRequestDeadlineToGatewayException(t *testing.T) { handler := HandlerFunc(func(ctx context.Context, req *Request) ([]byte, error) { <-ctx.Done() diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 7cf857f..b170e75 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -69,6 +69,29 @@ func (p *Proxy) Healthy() error { return p.client.Healthy() } +// HealthStatus returns optional upstream reliability diagnostics. +func (p *Proxy) HealthStatus() (string, map[string]any) { + reporter, ok := p.client.(interface { + HealthStatus() (string, map[string]any) + }) + if !ok { + return "ok", nil + } + return reporter.HealthStatus() +} + +// HealthReport returns reliability diagnostics and availability from one snapshot. +func (p *Proxy) HealthReport() (string, map[string]any, error) { + reporter, ok := p.client.(interface { + HealthReport() (string, map[string]any, error) + }) + if ok { + return reporter.HealthReport() + } + status, details := p.HealthStatus() + return status, details, p.Healthy() +} + // Run starts the proxy server. func (p *Proxy) Run(ctx context.Context) error { p.logger.Info("starting proxy", @@ -164,7 +187,7 @@ func (p *Proxy) handleRead(ctx context.Context, req *modbus.Request) ([]byte, er ) 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) { + data, coalesced, err := p.cache.CoalesceWithStatus(ctx, rangeKey, func(ctx context.Context) ([]byte, error) { data, err := p.client.Execute(ctx, req) if err != nil { return nil, err @@ -182,24 +205,124 @@ func (p *Proxy) handleRead(ctx context.Context, req *modbus.Request) ([]byte, er if err != nil { // Try serving stale data if configured - if p.cfg.CacheServeStale && modbus.ErrorKindOf(err) != modbus.ErrorProtocolException { + if p.cfg.CacheServeStale && + modbus.ErrorKindOf(err) != modbus.ErrorProtocolException && + requestContextActive(ctx) { 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", + if coalesced.Coalesced { + p.logCoalescedFailure(req, coalesced, err, true) + } else if coalesced.Waited { + p.logCoalescingWait(req, coalesced, err) + } + p.logger.Warn("stale fallback selected", "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", false, + "coalesced", coalesced.Coalesced, + "coalesced_waited", coalesced.Waited, + "coalesced_wait_duration", coalesced.WaitDuration, + "stale_fallback_selected", true, + "error_kind", modbus.ErrorKindOf(err), "error", err, ) return assembleResponse(req.FunctionCode, req.Quantity, staleValues), nil } } + if coalesced.Coalesced { + p.logCoalescedFailure(req, coalesced, err, false) + } else if coalesced.Waited { + p.logCoalescingWait(req, coalesced, err) + } return nil, err } + if coalesced.Coalesced { + p.logCoalescedCompletion(req, coalesced) + } else if coalesced.Waited { + p.logCoalescingWait(req, coalesced, nil) + } + return data, nil } +func requestContextActive(ctx context.Context) bool { + if ctx.Err() != nil { + return false + } + deadline, ok := ctx.Deadline() + return !ok || time.Now().Before(deadline) +} + +func (p *Proxy) logCoalescedCompletion(req *modbus.Request, result cache.CoalesceResult) { + p.logger.Debug("coalesced request completed", + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", false, + "attempt", 0, + "attempts", 0, + "queue_duration", result.WaitDuration, + "attempt_duration", time.Duration(0), + "reconnect_duration", time.Duration(0), + "total_duration", result.WaitDuration, + "error_kind", "", + "will_retry", false, + "coalesced", true, + "coalesced_waited", true, + "coalesced_wait_duration", result.WaitDuration, + ) +} + +func (p *Proxy) logCoalescedFailure(req *modbus.Request, result cache.CoalesceResult, err error, staleFallbackSelected bool) { + args := []any{ + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", false, + "attempt", 0, + "attempts", 0, + "queue_duration", result.WaitDuration, + "attempt_duration", time.Duration(0), + "reconnect_duration", time.Duration(0), + "total_duration", result.WaitDuration, + "error_kind", modbus.ErrorKindOf(err), + "will_retry", false, + "coalesced", true, + "coalesced_waited", true, + "coalesced_wait_duration", result.WaitDuration, + "stale_fallback_selected", staleFallbackSelected, + "error", err, + } + if !staleFallbackSelected { + args = append(args, "downstream_exception", fmt.Sprintf("0x%02X", modbus.DownstreamException(err))) + } + p.logger.Debug("coalesced request failed", args...) +} + +func (p *Proxy) logCoalescingWait(req *modbus.Request, result cache.CoalesceResult, err error) { + args := []any{ + "slave_id", req.SlaveID, + "func", fmt.Sprintf("0x%02X", req.FunctionCode), + "addr", req.Address, + "qty", req.Quantity, + "write", false, + "coalesced", false, + "coalesced_waited", true, + "coalesced_wait_duration", result.WaitDuration, + } + if err != nil { + args = append(args, "error_kind", modbus.ErrorKindOf(err), "error", err) + } + p.logger.Debug("coalescing wait ended before replacement fetch", args...) +} + func (p *Proxy) handleWrite(ctx context.Context, req *modbus.Request) ([]byte, error) { switch p.cfg.ReadOnly { case config.ReadOnlyOn: diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index a46f4a1..bb84346 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -3,6 +3,7 @@ package proxy import ( "bytes" "context" + "encoding/json" "errors" "io" "log/slog" @@ -22,6 +23,18 @@ type mockClient struct { calls int } +type healthReportingClient struct { + mockClient +} + +func (c *healthReportingClient) HealthStatus() (string, map[string]any) { + return "degraded", map[string]any{"total_retries": uint64(2)} +} + +func (c *healthReportingClient) HealthReport() (string, map[string]any, error) { + return "degraded", map[string]any{"total_retries": uint64(2)}, nil +} + type interleavingClient struct { mu sync.Mutex executeMu sync.Mutex @@ -102,6 +115,125 @@ func (m *mockClient) Execute(ctx context.Context, req *modbus.Request) ([]byte, return m.response, m.err } +func TestProxy_DelegatesHealthDiagnostics(t *testing.T) { + p := &Proxy{client: &healthReportingClient{}} + status, details := p.HealthStatus() + if status != "degraded" || details["total_retries"] != uint64(2) { + t.Fatalf("status=%q details=%+v", status, details) + } + status, details, err := p.HealthReport() + if err != nil || status != "degraded" || details["total_retries"] != uint64(2) { + t.Fatalf("atomic status=%q details=%+v err=%v", status, details, err) + } +} + +func TestProxy_CoalescedCompletionLogsFollowerAttribution(t *testing.T) { + var output bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})) + p := &Proxy{logger: logger} + req := &modbus.Request{ + SlaveID: 3, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 120, + Quantity: 4, + } + + p.logCoalescedCompletion(req, cache.CoalesceResult{ + Coalesced: true, + Waited: true, + WaitDuration: 5 * time.Second, + }) + + var entry map[string]any + if err := json.Unmarshal(bytes.TrimSpace(output.Bytes()), &entry); err != nil { + t.Fatalf("decode log: %v", err) + } + for _, field := range []string{ + "slave_id", "func", "addr", "qty", "write", "attempt", "attempts", + "queue_duration", "attempt_duration", "reconnect_duration", + "total_duration", "error_kind", "will_retry", "coalesced", + "coalesced_wait_duration", + } { + if _, ok := entry[field]; !ok { + t.Errorf("log entry missing %q: %+v", field, entry) + } + } + if entry["attempts"] != float64(0) || + entry["attempt_duration"] != float64(0) || + entry["reconnect_duration"] != float64(0) || + entry["coalesced"] != true || + entry["queue_duration"] != float64(5*time.Second) { + t.Fatalf("coalesced follower claimed an upstream attempt: %+v", entry) + } +} + +func TestProxy_CoalescedFailureLogsWaitWithoutAttempt(t *testing.T) { + var output bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})) + p := &Proxy{logger: logger} + req := &modbus.Request{ + SlaveID: 3, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 120, + Quantity: 4, + } + err := &modbus.RequestError{ + Kind: modbus.ErrorTransportTimeout, + Attempts: 2, + Err: context.DeadlineExceeded, + } + + p.logCoalescedFailure(req, cache.CoalesceResult{ + Coalesced: true, + Waited: true, + WaitDuration: 5 * time.Second, + }, err, false) + + var entry map[string]any + if err := json.Unmarshal(bytes.TrimSpace(output.Bytes()), &entry); err != nil { + t.Fatalf("decode log: %v", err) + } + if entry["attempts"] != float64(0) || + entry["attempt_duration"] != float64(0) || + entry["reconnect_duration"] != float64(0) || + entry["coalesced"] != true || + entry["error_kind"] != string(modbus.ErrorTransportTimeout) || + entry["stale_fallback_selected"] != false || + entry["downstream_exception"] != "0x0B" { + t.Fatalf("coalesced failure claimed leader diagnostics: %+v", entry) + } +} + +func TestProxy_ReplacementLeaderLogsPriorCoalescingWait(t *testing.T) { + var output bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})) + p := &Proxy{logger: logger} + req := &modbus.Request{ + SlaveID: 3, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 120, + Quantity: 4, + } + + p.logCoalescingWait(req, cache.CoalesceResult{ + Waited: true, + WaitDuration: 5 * time.Second, + }, nil) + + var entry map[string]any + if err := json.Unmarshal(bytes.TrimSpace(output.Bytes()), &entry); err != nil { + t.Fatalf("decode log: %v", err) + } + if entry["coalesced"] != false || + entry["coalesced_waited"] != true || + entry["coalesced_wait_duration"] != float64(5*time.Second) { + t.Fatalf("replacement leader lost wait attribution: %+v", entry) + } + if _, ok := entry["attempts"]; ok { + t.Fatalf("wait-only event claimed an upstream attempt: %+v", entry) + } +} + func TestProxy_HandleReadCacheHit(t *testing.T) { logger := slog.New(slog.NewTextHandler(io.Discard, nil)) c := cache.New(time.Second, false) @@ -210,7 +342,8 @@ func TestProxy_HandleReadMissFetchesAndCaches(t *testing.T) { } func TestProxy_HandleReadServesStaleOnUpstreamError(t *testing.T) { - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + var output bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})) c := cache.New(10*time.Millisecond, true) defer c.Close() @@ -252,6 +385,79 @@ func TestProxy_HandleReadServesStaleOnUpstreamError(t *testing.T) { if !bytes.Equal(resp, expected) { t.Fatalf("stale response: expected %v, got %v", expected, resp) } + var entry map[string]any + for _, line := range bytes.Split(bytes.TrimSpace(output.Bytes()), []byte{'\n'}) { + var candidate map[string]any + if err := json.Unmarshal(line, &candidate); err != nil { + t.Fatalf("decode stale fallback log: %v", err) + } + if candidate["msg"] == "stale fallback selected" { + entry = candidate + break + } + } + if entry == nil || entry["stale_fallback_selected"] != true { + t.Fatalf("stale fallback selection not logged accurately: %+v", entry) + } +} + +func TestProxy_HandleReadDoesNotServeStaleAfterContextEnds(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, func() {} + }, + wantErr: context.Canceled, + }, + { + name: "deadline", + 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) { + var output bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})) + c := cache.New(time.Nanosecond, true) + defer c.Close() + c.SetRange(1, modbus.FuncReadHoldingRegisters, 20, [][]byte{{0x00, 0x01}}) + time.Sleep(time.Millisecond) + p := &Proxy{ + cfg: &config.Config{ + CacheServeStale: true, + ReadOnly: config.ReadOnlyOn, + }, + logger: logger, + client: &mockClient{err: errors.New("upstream unavailable")}, + cache: c, + } + req := &modbus.Request{ + SlaveID: 1, + FunctionCode: modbus.FuncReadHoldingRegisters, + Address: 20, + Quantity: 1, + } + ctx, cancel := tt.context() + defer cancel() + + if _, err := p.HandleRequest(ctx, req); !errors.Is(err, tt.wantErr) { + t.Fatalf("error = %v, want %v", err, tt.wantErr) + } + if bytes.Contains(output.Bytes(), []byte("stale fallback selected")) { + t.Fatalf("context-ended request selected stale fallback: %s", output.String()) + } + }) + } } func TestProxy_HandleReadDoesNotHideProtocolExceptionWithStaleData(t *testing.T) {