From a7076cfb4a4fa8c021a884702f119b17ca110da0 Mon Sep 17 00:00:00 2001 From: Souyama Date: Sun, 14 Jun 2026 19:19:04 +0530 Subject: [PATCH] Add backoff based recovery --- ascii_over_tcp_client.go | 15 ++ ascii_transport_test.go | 16 +- asciiclient.go | 14 +- backoff.go | 203 +++++++++++++++++++ backoff_test.go | 415 +++++++++++++++++++++++++++++++++++++++ rtu_over_tcp_client.go | 15 ++ rtu_transport_test.go | 16 +- rtuclient.go | 15 +- serial.go | 76 +++++-- serial_test.go | 197 ++++++++++++++++++- tcpclient.go | 130 +++++++++++- tcpclient_test.go | 219 +++++++++++++++++++++ 12 files changed, 1269 insertions(+), 62 deletions(-) create mode 100644 backoff.go create mode 100644 backoff_test.go diff --git a/ascii_over_tcp_client.go b/ascii_over_tcp_client.go index 8f0b926..737b177 100644 --- a/ascii_over_tcp_client.go +++ b/ascii_over_tcp_client.go @@ -16,12 +16,27 @@ type ASCIIOverTCPClientHandler struct { } // NewASCIIOverTCPClientHandler allocates and initializes a ASCIIOverTCPClientHandler. +// The handler uses exponential backoff (10ms-5s) with 30s timeout by default for link recovery. +// This is appropriate for TCP network links. For custom backoff, set LinkRecoveryBackoff explicitly. func NewASCIIOverTCPClientHandler(address string) *ASCIIOverTCPClientHandler { handler := &ASCIIOverTCPClientHandler{} handler.Address = address handler.Timeout = tcpTimeout handler.IdleTimeout = tcpIdleTimeout handler.Dial = defaultDialFunc(handler.Timeout) + // Default exponential backoff for TCP: 10ms initial, suitable for faster network recovery + handler.LinkRecoveryBackoff = NewExponentialBackoff( + 10*time.Millisecond, // Initial interval (network-appropriate) + 5*time.Second, // Max interval + 30*time.Second, // Timeout + ) + // Default protocol recovery for transaction ID mismatches: 10ms with 100ms timeout + // Protocol recovery is just processing junk data, fail fast if it persists + handler.ProtocolRecoveryBackoff = NewExponentialBackoff( + 10*time.Millisecond, // Initial interval + 50*time.Millisecond, // Max interval + 100*time.Millisecond, // Timeout (fail fast for junk data) + ) return handler } diff --git a/ascii_transport_test.go b/ascii_transport_test.go index 19c8cea..0f648e2 100644 --- a/ascii_transport_test.go +++ b/ascii_transport_test.go @@ -199,10 +199,10 @@ func TestASCIISerialTransporter_RecoveryDisabledOnReadEOF(t *testing.T) { _, err := transporter.Send(context.Background(), reqASCII) if err == nil { - t.Fatal("expected link recovery timeout error, got nil") + t.Fatal("expected link recovery error, got nil") } - if !strings.Contains(err.Error(), "link recovery timeout reached") || !errors.Is(err, io.EOF) { - t.Fatalf("expected link recovery timeout wrapping EOF, got %v", err) + if !strings.Contains(err.Error(), "no link recovery configured") || !errors.Is(err, io.EOF) { + t.Fatalf("expected no link recovery configured wrapping EOF, got %v", err) } if got := port.written.Bytes(); !bytes.Equal(got, reqASCII) { t.Fatalf("expected request %q, got %q", reqASCII, got) @@ -225,10 +225,10 @@ func TestASCIISerialTransporter_ReconnectBudgetExhaustedOnReadEOF(t *testing.T) _, err := transporter.Send(context.Background(), reqASCII) elapsed := time.Since(start) if err == nil { - t.Fatal("expected link recovery timeout error, got nil") + t.Fatal("expected link recovery exhausted error, got nil") } - if !strings.Contains(err.Error(), "link recovery timeout reached") { - t.Fatalf("expected link recovery timeout error, got %v", err) + if !strings.Contains(err.Error(), "link recovery exhausted") { + t.Fatalf("expected link recovery exhausted error, got %v", err) } if !strings.Contains(err.Error(), "could not open") { t.Fatalf("expected reconnect open failure to be wrapped, got %v", err) @@ -262,8 +262,8 @@ func TestASCIISerialTransporter_ReconnectOnWriteEOF(t *testing.T) { if err == nil { t.Fatal("expected reconnect error after write EOF, got nil") } - if !strings.Contains(err.Error(), "link recovery timeout reached") || !strings.Contains(err.Error(), "could not open") { - t.Fatalf("expected timed-out reconnect open failure, got %v", err) + if !strings.Contains(err.Error(), "link recovery exhausted") || !strings.Contains(err.Error(), "could not open") { + t.Fatalf("expected link recovery exhausted with reconnect failure, got %v", err) } if elapsed < recoveryTimeout-20*time.Millisecond { t.Fatalf("expected recovery to keep retrying for about %v, returned after %v", recoveryTimeout, elapsed) diff --git a/asciiclient.go b/asciiclient.go index 3aa6a0d..a921f86 100644 --- a/asciiclient.go +++ b/asciiclient.go @@ -31,12 +31,18 @@ type ASCIIClientHandler struct { } // NewASCIIClientHandler allocates and initializes a ASCIIClientHandler. +// The handler uses exponential backoff (100ms-5s) with 30s timeout by default for link recovery. func NewASCIIClientHandler(address string) *ASCIIClientHandler { handler := &ASCIIClientHandler{} handler.Address = address handler.Timeout = serialTimeout handler.IdleTimeout = serialIdleTimeout - handler.ReconnectRetryInterval = serialReconnectRetryInterval + // Default exponential backoff for RS485: 100ms→2s with 30s timeout + handler.LinkRecoveryBackoff = NewExponentialBackoff( + 100*time.Millisecond, // Initial interval (RS485-appropriate) + 5*time.Second, // Max interval cap + 30*time.Second, // Timeout + ) return handler } @@ -183,14 +189,12 @@ func (mb *asciiSerialTransporter) Send(ctx context.Context, aduRequest []byte) ( mb.lastActivity = time.Now() mb.startCloseTimer() - linkRecoveryDeadline := time.Now().Add(mb.LinkRecoveryTimeout) - for { // Send the request mb.logf("modbus: send % x\n", aduRequest) if _, err = mb.port.Write(aduRequest); err != nil { if mb.shouldRecover(err) { - if err = mb.reconnect(ctx, err, linkRecoveryDeadline); err != nil { + if err = mb.reconnect(ctx, err); err != nil { return } continue @@ -206,7 +210,7 @@ func (mb *asciiSerialTransporter) Send(ctx context.Context, aduRequest []byte) ( } if err != nil { if mb.shouldRecover(err) { - if err = mb.reconnect(ctx, err, linkRecoveryDeadline); err != nil { + if err = mb.reconnect(ctx, err); err != nil { return } continue diff --git a/backoff.go b/backoff.go new file mode 100644 index 0000000..614ef5a --- /dev/null +++ b/backoff.go @@ -0,0 +1,203 @@ +package modbus + +import ( + "math" + "time" +) + +// BackoffType defines the retry interval growth algorithm +type BackoffType int + +const ( + BackoffFixed BackoffType = iota // Constant interval + BackoffLinear // Linear growth + BackoffExponential // Exponential growth +) + +// BackoffStrategy defines a unified retry strategy for connection recovery. +// It controls retry intervals, total timeout budget, and maximum attempt count. +// +// Example usage with exponential backoff and timeout: +// +// handler := modbus.NewRTUClientHandler("/dev/ttyUSB0") +// handler.LinkRecoveryBackoff = &modbus.BackoffStrategy{ +// Type: modbus.BackoffExponential, +// InitialInterval: 10 * time.Millisecond, +// MaxInterval: 5 * time.Second, +// Multiplier: 2.0, +// Timeout: 30 * time.Second, +// MaxAttempts: 0, // unlimited attempts within timeout +// } +// // Retry progression: 10ms → 20ms → 40ms → 80ms → ... → 5s (capped) +// +// Example with max attempts: +// +// handler.LinkRecoveryBackoff = &modbus.BackoffStrategy{ +// Type: modbus.BackoffFixed, +// InitialInterval: 100 * time.Millisecond, +// MaxInterval: 100 * time.Millisecond, +// Timeout: 0, // unlimited time +// MaxAttempts: 10, +// } +type BackoffStrategy struct { + // Type specifies the interval growth algorithm + Type BackoffType + // InitialInterval is the starting delay for the first retry attempt + InitialInterval time.Duration + // MaxInterval caps the maximum delay to prevent unbounded growth (0 = unlimited) + MaxInterval time.Duration + // Multiplier is the growth factor for exponential/linear backoff + // - Exponential: interval = InitialInterval * (Multiplier ^ attempt) + // - Linear: interval = InitialInterval * (1 + attempt * Multiplier) + // - Fixed: Multiplier is ignored + Multiplier float64 + + // Timeout is the total time budget for all retry attempts (0 = unlimited) + Timeout time.Duration + // MaxAttempts is the maximum number of retry attempts (0 = unlimited) + MaxAttempts int +} + +// Next calculates the backoff interval for the given attempt number. +// attempt is zero-indexed (0 = first retry, 1 = second retry, etc.) +// The returned duration is capped at MaxInterval if MaxInterval > 0, +// otherwise growth is unlimited. +func (b *BackoffStrategy) Next(attempt int) time.Duration { + if b == nil { + return 0 + } + + var interval time.Duration + + switch b.Type { + case BackoffFixed: + interval = b.InitialInterval + + case BackoffLinear: + // Linear: InitialInterval * (1 + attempt * Multiplier) + multiplier := b.Multiplier + if multiplier <= 0 { + multiplier = 1.0 // Default linear growth + } + factor := 1.0 + float64(attempt)*multiplier + interval = time.Duration(float64(b.InitialInterval) * factor) + + case BackoffExponential: + // Exponential: InitialInterval * (Multiplier ^ attempt) + multiplier := b.Multiplier + if multiplier <= 1.0 { + multiplier = 2.0 // Default exponential growth + } + factor := math.Pow(multiplier, float64(attempt)) + interval = time.Duration(float64(b.InitialInterval) * factor) + + default: + // Default to fixed if type is unrecognized + interval = b.InitialInterval + } + + // Cap at MaxInterval + if b.MaxInterval > 0 && interval > b.MaxInterval { + interval = b.MaxInterval + } + + return interval +} + +// ShouldRetry determines whether another retry attempt should be made. +// It checks both the timeout and max attempts conditions. +// Returns false if either limit has been reached, true otherwise. +func (b *BackoffStrategy) ShouldRetry(attempt int, elapsed time.Duration) bool { + if b == nil { + return false + } + + // Check timeout limit + if b.Timeout > 0 && elapsed >= b.Timeout { + return false + } + + // Check max attempts limit + if b.MaxAttempts > 0 && attempt >= b.MaxAttempts { + return false + } + + return true +} + +// Clone creates an independent copy of the BackoffStrategy. +func (b *BackoffStrategy) Clone() *BackoffStrategy { + if b == nil { + return nil + } + return &BackoffStrategy{ + Type: b.Type, + InitialInterval: b.InitialInterval, + MaxInterval: b.MaxInterval, + Multiplier: b.Multiplier, + Timeout: b.Timeout, + MaxAttempts: b.MaxAttempts, + } +} + +// NewExponentialBackoff returns an exponential backoff strategy with custom intervals and timeout. +// Progression: initial → initial*2 → initial*4 → ... → max (capped) +func NewExponentialBackoff(initialInterval, maxInterval, timeout time.Duration) *BackoffStrategy { + return &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: initialInterval, + MaxInterval: maxInterval, + Multiplier: 2.0, + Timeout: timeout, + MaxAttempts: 0, + } +} + +// NewLinearBackoff returns a linear backoff strategy with custom intervals and timeout. +// Progression: initial → initial*2 → initial*3 → ... → max (capped) +func NewLinearBackoff(initialInterval, maxInterval, timeout time.Duration) *BackoffStrategy { + return &BackoffStrategy{ + Type: BackoffLinear, + InitialInterval: initialInterval, + MaxInterval: maxInterval, + Multiplier: 1.0, + Timeout: timeout, + MaxAttempts: 0, + } +} + +// NewFixedBackoff returns a backoff strategy with a constant retry interval. +func NewFixedBackoff(interval, timeout time.Duration) *BackoffStrategy { + return &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: interval, + MaxInterval: interval, + Multiplier: 0, + Timeout: timeout, + MaxAttempts: 0, + } +} + +// NewLinearBackoffWithMaxAttempts returns a linear backoff strategy with max attempt limit instead of timeout. +func NewLinearBackoffWithMaxAttempts(initialInterval, maxInterval time.Duration, maxAttempts int) *BackoffStrategy { + return &BackoffStrategy{ + Type: BackoffLinear, + InitialInterval: initialInterval, + MaxInterval: maxInterval, + Multiplier: 1.0, + Timeout: 0, + MaxAttempts: maxAttempts, + } +} + +// NewExponentialBackoffWithMaxAttempts returns an exponential backoff strategy with max attempt limit instead of timeout. +func NewExponentialBackoffWithMaxAttempts(initialInterval, maxInterval time.Duration, maxAttempts int) *BackoffStrategy { + return &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: initialInterval, + MaxInterval: maxInterval, + Multiplier: 2.0, + Timeout: 0, + MaxAttempts: maxAttempts, + } +} diff --git a/backoff_test.go b/backoff_test.go new file mode 100644 index 0000000..3cb4000 --- /dev/null +++ b/backoff_test.go @@ -0,0 +1,415 @@ +package modbus + +import ( + "testing" + "time" +) + +func TestBackoffStrategy_Exponential(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 5 * time.Second, + Multiplier: 2.0, + } + + // Test exponential progression: 10ms, 20ms, 40ms, 80ms, ... + expected := []time.Duration{ + 10 * time.Millisecond, // attempt 0: 10 * 2^0 = 10 + 20 * time.Millisecond, // attempt 1: 10 * 2^1 = 20 + 40 * time.Millisecond, // attempt 2: 10 * 2^2 = 40 + 80 * time.Millisecond, // attempt 3: 10 * 2^3 = 80 + 160 * time.Millisecond, // attempt 4: 10 * 2^4 = 160 + 320 * time.Millisecond, // attempt 5: 10 * 2^5 = 320 + 640 * time.Millisecond, // attempt 6: 10 * 2^6 = 640 + 1280 * time.Millisecond, // attempt 7: 10 * 2^7 = 1280 + 2560 * time.Millisecond, // attempt 8: 10 * 2^8 = 2560 + 5 * time.Second, // attempt 9: 10 * 2^9 = 5120ms, capped at 5s + 5 * time.Second, // attempt 10: capped at 5s + } + + for i, want := range expected { + got := strategy.Next(i) + if got != want { + t.Errorf("attempt %d: got %v, want %v", i, got, want) + } + } +} + +func TestBackoffStrategy_Linear(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffLinear, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 100 * time.Millisecond, + Multiplier: 1.0, + } + + // Test linear progression: 10ms, 20ms, 30ms, 40ms, ... + expected := []time.Duration{ + 10 * time.Millisecond, // attempt 0: 10 * (1 + 0*1.0) = 10 + 20 * time.Millisecond, // attempt 1: 10 * (1 + 1*1.0) = 20 + 30 * time.Millisecond, // attempt 2: 10 * (1 + 2*1.0) = 30 + 40 * time.Millisecond, // attempt 3: 10 * (1 + 3*1.0) = 40 + 50 * time.Millisecond, // attempt 4: 10 * (1 + 4*1.0) = 50 + 60 * time.Millisecond, // attempt 5: 10 * (1 + 5*1.0) = 60 + 70 * time.Millisecond, // attempt 6: 10 * (1 + 6*1.0) = 70 + 80 * time.Millisecond, // attempt 7: 10 * (1 + 7*1.0) = 80 + 90 * time.Millisecond, // attempt 8: 10 * (1 + 8*1.0) = 90 + 100 * time.Millisecond, // attempt 9: 10 * (1 + 9*1.0) = 100, at cap + 100 * time.Millisecond, // attempt 10: capped at 100ms + } + + for i, want := range expected { + got := strategy.Next(i) + if got != want { + t.Errorf("attempt %d: got %v, want %v", i, got, want) + } + } +} + +func TestBackoffStrategy_Fixed(t *testing.T) { + interval := 100 * time.Millisecond + strategy := &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: interval, + MaxInterval: interval, + } + + // All attempts should return the same interval + for i := 0; i < 10; i++ { + got := strategy.Next(i) + if got != interval { + t.Errorf("attempt %d: got %v, want %v", i, got, interval) + } + } +} + +func TestBackoffStrategy_ShouldRetry_Timeout(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 10 * time.Millisecond, + Timeout: 100 * time.Millisecond, + MaxAttempts: 0, // unlimited attempts + } + + // Should allow retries within timeout + if !strategy.ShouldRetry(0, 0) { + t.Error("should retry at start") + } + if !strategy.ShouldRetry(5, 50*time.Millisecond) { + t.Error("should retry at 50ms (within 100ms timeout)") + } + if !strategy.ShouldRetry(10, 99*time.Millisecond) { + t.Error("should retry at 99ms (just under timeout)") + } + + // Should not retry after timeout + if strategy.ShouldRetry(15, 100*time.Millisecond) { + t.Error("should not retry at exact timeout") + } + if strategy.ShouldRetry(20, 150*time.Millisecond) { + t.Error("should not retry past timeout") + } +} + +func TestBackoffStrategy_ShouldRetry_MaxAttempts(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 10 * time.Millisecond, + Timeout: 0, // unlimited time + MaxAttempts: 5, + } + + // Should allow retries up to max attempts + if !strategy.ShouldRetry(0, 0) { + t.Error("should retry attempt 0") + } + if !strategy.ShouldRetry(4, time.Second) { + t.Error("should retry attempt 4 (just under max)") + } + + // Should not retry at or past max attempts + if strategy.ShouldRetry(5, time.Second) { + t.Error("should not retry at max attempts") + } + if strategy.ShouldRetry(10, time.Second) { + t.Error("should not retry past max attempts") + } +} + +func TestBackoffStrategy_ShouldRetry_Combined(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 10 * time.Millisecond, + Timeout: 100 * time.Millisecond, + MaxAttempts: 10, + } + + // Should retry when both conditions are met + if !strategy.ShouldRetry(5, 50*time.Millisecond) { + t.Error("should retry when both within limits") + } + + // Should stop when timeout is reached (even if attempts under limit) + if strategy.ShouldRetry(8, 100*time.Millisecond) { + t.Error("should stop when timeout reached") + } + + // Should stop when max attempts reached (even if time under limit) + if strategy.ShouldRetry(10, 50*time.Millisecond) { + t.Error("should stop when max attempts reached") + } +} + +func TestBackoffStrategy_ShouldRetry_Unlimited(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 10 * time.Millisecond, + Timeout: 0, // unlimited + MaxAttempts: 0, // unlimited + } + + // Should always retry when both are unlimited + if !strategy.ShouldRetry(0, 0) { + t.Error("should retry at start") + } + if !strategy.ShouldRetry(1000, time.Hour) { + t.Error("should retry even after long time and many attempts") + } +} + +func TestBackoffStrategy_NilHandling(t *testing.T) { + var strategy *BackoffStrategy + + got := strategy.Next(0) + if got != 0 { + t.Errorf("nil strategy Next() should return 0, got %v", got) + } + + if strategy.ShouldRetry(0, 0) { + t.Error("nil strategy ShouldRetry() should return false") + } + + cloned := strategy.Clone() + if cloned != nil { + t.Errorf("cloning nil strategy should return nil, got %v", cloned) + } +} + +func TestBackoffStrategy_Clone(t *testing.T) { + original := &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 5 * time.Second, + Multiplier: 2.0, + Timeout: 30 * time.Second, + MaxAttempts: 10, + } + + cloned := original.Clone() + + // Verify all fields are copied + if cloned.Type != original.Type { + t.Errorf("Type: got %v, want %v", cloned.Type, original.Type) + } + if cloned.InitialInterval != original.InitialInterval { + t.Errorf("InitialInterval: got %v, want %v", cloned.InitialInterval, original.InitialInterval) + } + if cloned.MaxInterval != original.MaxInterval { + t.Errorf("MaxInterval: got %v, want %v", cloned.MaxInterval, original.MaxInterval) + } + if cloned.Multiplier != original.Multiplier { + t.Errorf("Multiplier: got %v, want %v", cloned.Multiplier, original.Multiplier) + } + if cloned.Timeout != original.Timeout { + t.Errorf("Timeout: got %v, want %v", cloned.Timeout, original.Timeout) + } + if cloned.MaxAttempts != original.MaxAttempts { + t.Errorf("MaxAttempts: got %v, want %v", cloned.MaxAttempts, original.MaxAttempts) + } + + // Verify independence: modifying clone doesn't affect original + cloned.Type = BackoffLinear + cloned.Multiplier = 3.0 + cloned.Timeout = time.Hour + cloned.MaxAttempts = 100 + + if original.Type == cloned.Type { + t.Error("modifying clone affected original Type") + } + if original.Multiplier == cloned.Multiplier { + t.Error("modifying clone affected original Multiplier") + } + if original.Timeout == cloned.Timeout { + t.Error("modifying clone affected original Timeout") + } + if original.MaxAttempts == cloned.MaxAttempts { + t.Error("modifying clone affected original MaxAttempts") + } +} + +func TestNewExponentialBackoff(t *testing.T) { + strategy := NewExponentialBackoff(10*time.Millisecond, 5*time.Second, 30*time.Second) + + if strategy.Type != BackoffExponential { + t.Errorf("Type: got %v, want BackoffExponential", strategy.Type) + } + if strategy.InitialInterval != 10*time.Millisecond { + t.Errorf("InitialInterval: got %v, want 10ms", strategy.InitialInterval) + } + if strategy.MaxInterval != 5*time.Second { + t.Errorf("MaxInterval: got %v, want 5s", strategy.MaxInterval) + } + if strategy.Multiplier != 2.0 { + t.Errorf("Multiplier: got %v, want 2.0", strategy.Multiplier) + } + if strategy.Timeout != 30*time.Second { + t.Errorf("Timeout: got %v, want 30s", strategy.Timeout) + } + if strategy.MaxAttempts != 0 { + t.Errorf("MaxAttempts: got %v, want 0 (unlimited)", strategy.MaxAttempts) + } + + // Verify it produces expected progression + first := strategy.Next(0) + second := strategy.Next(1) + if first != 10*time.Millisecond { + t.Errorf("first attempt: got %v, want 10ms", first) + } + if second != 20*time.Millisecond { + t.Errorf("second attempt: got %v, want 20ms", second) + } +} + +func TestNewLinearBackoff(t *testing.T) { + strategy := NewLinearBackoff(10*time.Millisecond, 5*time.Second, 30*time.Second) + + if strategy.Type != BackoffLinear { + t.Errorf("Type: got %v, want BackoffLinear", strategy.Type) + } + if strategy.InitialInterval != 10*time.Millisecond { + t.Errorf("InitialInterval: got %v, want 10ms", strategy.InitialInterval) + } + if strategy.MaxInterval != 5*time.Second { + t.Errorf("MaxInterval: got %v, want 5s", strategy.MaxInterval) + } + if strategy.Timeout != 30*time.Second { + t.Errorf("Timeout: got %v, want 30s", strategy.Timeout) + } +} + +func TestNewFixedBackoff(t *testing.T) { + interval := 250 * time.Millisecond + timeout := 10 * time.Second + strategy := NewFixedBackoff(interval, timeout) + + if strategy.Type != BackoffFixed { + t.Errorf("Type: got %v, want BackoffFixed", strategy.Type) + } + if strategy.Timeout != timeout { + t.Errorf("Timeout: got %v, want %v", strategy.Timeout, timeout) + } + + // Verify it returns constant interval + for i := 0; i < 5; i++ { + got := strategy.Next(i) + if got != interval { + t.Errorf("attempt %d: got %v, want %v", i, got, interval) + } + } +} + +func TestNewFixedBackoff_NoTimeout(t *testing.T) { + interval := 200 * time.Millisecond + strategy := NewFixedBackoff(interval, 0) + + if strategy.Type != BackoffFixed { + t.Errorf("Type: got %v, want BackoffFixed", strategy.Type) + } + if strategy.Timeout != 0 { + t.Errorf("Timeout: got %v, want 0 (unlimited)", strategy.Timeout) + } + + // Verify it returns constant interval + for i := 0; i < 5; i++ { + got := strategy.Next(i) + if got != interval { + t.Errorf("attempt %d: got %v, want %v", i, got, interval) + } + } +} + +func TestNewLinearBackoffWithMaxAttempts(t *testing.T) { + strategy := NewLinearBackoffWithMaxAttempts(100*time.Millisecond, 1*time.Second, 20) + + if strategy.Type != BackoffLinear { + t.Errorf("Type: got %v, want BackoffLinear", strategy.Type) + } + if strategy.MaxAttempts != 20 { + t.Errorf("MaxAttempts: got %v, want 20", strategy.MaxAttempts) + } + if strategy.Timeout != 0 { + t.Errorf("Timeout: got %v, want 0 (unlimited)", strategy.Timeout) + } +} + +func TestNewExponentialBackoffWithMaxAttempts(t *testing.T) { + strategy := NewExponentialBackoffWithMaxAttempts(50*time.Millisecond, 2*time.Second, 10) + + if strategy.Type != BackoffExponential { + t.Errorf("Type: got %v, want BackoffExponential", strategy.Type) + } + if strategy.MaxAttempts != 10 { + t.Errorf("MaxAttempts: got %v, want 10", strategy.MaxAttempts) + } + if strategy.Timeout != 0 { + t.Errorf("Timeout: got %v, want 0 (unlimited)", strategy.Timeout) + } +} + +func TestBackoffType_InvalidType(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffType(99), // Invalid type + InitialInterval: 50 * time.Millisecond, + MaxInterval: 100 * time.Millisecond, + } + + // Should default to fixed behavior for unknown types + for i := 0; i < 3; i++ { + got := strategy.Next(i) + if got != 50*time.Millisecond { + t.Errorf("attempt %d: got %v, want 50ms (fixed fallback)", i, got) + } + } +} + +func TestBackoffStrategy_UnlimitedMaxInterval(t *testing.T) { + strategy := &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 0, // Unlimited growth + Multiplier: 2.0, + Timeout: 0, + MaxAttempts: 0, + } + + // Verify exponential growth continues without cap + expected := []time.Duration{ + 10 * time.Millisecond, // 10ms * 2^0 + 20 * time.Millisecond, // 10ms * 2^1 + 40 * time.Millisecond, // 10ms * 2^2 + 80 * time.Millisecond, // 10ms * 2^3 + 160 * time.Millisecond, // 10ms * 2^4 + 320 * time.Millisecond, // 10ms * 2^5 + 640 * time.Millisecond, // 10ms * 2^6 + 1280 * time.Millisecond, // 10ms * 2^7 + 2560 * time.Millisecond, // 10ms * 2^8 + } + + for i, want := range expected { + got := strategy.Next(i) + if got != want { + t.Errorf("attempt %d: got %v, want %v", i, got, want) + } + } +} diff --git a/rtu_over_tcp_client.go b/rtu_over_tcp_client.go index b2c2e0c..6311e8b 100644 --- a/rtu_over_tcp_client.go +++ b/rtu_over_tcp_client.go @@ -17,12 +17,27 @@ type RTUOverTCPClientHandler struct { } // NewRTUOverTCPClientHandler allocates and initializes a RTUOverTCPClientHandler. +// The handler uses exponential backoff (10ms-5s) with 30s timeout by default for link recovery. +// This is appropriate for TCP network links. For custom backoff, set LinkRecoveryBackoff explicitly. func NewRTUOverTCPClientHandler(address string) *RTUOverTCPClientHandler { handler := &RTUOverTCPClientHandler{} handler.Address = address handler.Timeout = tcpTimeout handler.IdleTimeout = tcpIdleTimeout handler.Dial = defaultDialFunc(handler.Timeout) + // Default exponential backoff for TCP: 10ms initial, suitable for faster network recovery + handler.LinkRecoveryBackoff = NewExponentialBackoff( + 10*time.Millisecond, // Initial interval (network-appropriate) + 5*time.Second, // Max interval + 30*time.Second, // Timeout + ) + // Default protocol recovery for transaction ID mismatches: 10ms with 100ms timeout + // Protocol recovery is just processing junk data, fail fast if it persists + handler.ProtocolRecoveryBackoff = NewExponentialBackoff( + 10*time.Millisecond, // Initial interval + 50*time.Millisecond, // Max interval + 100*time.Millisecond, // Timeout (fail fast for junk data) + ) return handler } diff --git a/rtu_transport_test.go b/rtu_transport_test.go index 02d480b..632cb80 100644 --- a/rtu_transport_test.go +++ b/rtu_transport_test.go @@ -271,10 +271,10 @@ func TestRTUSerialTransporter_RecoveryDisabledOnReadEOF(t *testing.T) { _, err := transporter.Send(context.Background(), req) if err == nil { - t.Fatal("expected link recovery timeout error, got nil") + t.Fatal("expected link recovery error, got nil") } - if !strings.Contains(err.Error(), "link recovery timeout reached") || !errors.Is(err, io.EOF) { - t.Fatalf("expected link recovery timeout wrapping EOF, got %v", err) + if !strings.Contains(err.Error(), "no link recovery configured") || !errors.Is(err, io.EOF) { + t.Fatalf("expected no link recovery configured wrapping EOF, got %v", err) } if got := port.written.Bytes(); !bytes.Equal(got, req) { t.Fatalf("expected request %x, got %x", req, got) @@ -297,10 +297,10 @@ func TestRTUSerialTransporter_ReconnectBudgetExhaustedOnReadEOF(t *testing.T) { _, err := transporter.Send(context.Background(), req) elapsed := time.Since(start) if err == nil { - t.Fatal("expected link recovery timeout error, got nil") + t.Fatal("expected link recovery exhausted error, got nil") } - if !strings.Contains(err.Error(), "link recovery timeout reached") { - t.Fatalf("expected link recovery timeout error, got %v", err) + if !strings.Contains(err.Error(), "link recovery exhausted") { + t.Fatalf("expected link recovery exhausted error, got %v", err) } if !strings.Contains(err.Error(), "could not open") { t.Fatalf("expected reconnect open failure to be wrapped, got %v", err) @@ -334,8 +334,8 @@ func TestRTUSerialTransporter_ReconnectOnWriteEOF(t *testing.T) { if err == nil { t.Fatal("expected reconnect error after write EOF, got nil") } - if !strings.Contains(err.Error(), "link recovery timeout reached") || !strings.Contains(err.Error(), "could not open") { - t.Fatalf("expected timed-out reconnect open failure, got %v", err) + if !strings.Contains(err.Error(), "link recovery exhausted") || !strings.Contains(err.Error(), "could not open") { + t.Fatalf("expected link recovery exhausted with reconnect failure, got %v", err) } if elapsed < recoveryTimeout-20*time.Millisecond { t.Fatalf("expected recovery to keep retrying for about %v, returned after %v", recoveryTimeout, elapsed) diff --git a/rtuclient.go b/rtuclient.go index 9874ba4..5f072d7 100644 --- a/rtuclient.go +++ b/rtuclient.go @@ -50,12 +50,19 @@ type RTUClientHandler struct { } // NewRTUClientHandler allocates and initializes a RTUClientHandler. +// The handler uses exponential backoff (100ms-5s) with 30s timeout by default for link recovery. +// This is appropriate for RS485 serial links. For custom backoff, set LinkRecoveryBackoff explicitly. func NewRTUClientHandler(address string) *RTUClientHandler { handler := &RTUClientHandler{} handler.Address = address handler.Timeout = serialTimeout handler.IdleTimeout = serialIdleTimeout - handler.ReconnectRetryInterval = serialReconnectRetryInterval + // Default exponential backoff for RS485: 100ms→2s with 30s timeout + handler.LinkRecoveryBackoff = NewExponentialBackoff( + 100*time.Millisecond, // Initial interval (RS485-appropriate) + 5*time.Second, // Max interval cap + 30*time.Second, // Timeout + ) return handler } @@ -264,14 +271,12 @@ func (mb *rtuSerialTransporter) Send(ctx context.Context, aduRequest []byte) (ad mb.lastActivity = time.Now() mb.startCloseTimer() - linkRecoveryDeadline := time.Now().Add(mb.LinkRecoveryTimeout) - for { // Send the request mb.logf("modbus: send % x\n", aduRequest) if _, err = mb.port.Write(aduRequest); err != nil { if mb.shouldRecover(err) { - if err = mb.reconnect(ctx, err, linkRecoveryDeadline); err != nil { + if err = mb.reconnect(ctx, err); err != nil { return } continue @@ -294,7 +299,7 @@ func (mb *rtuSerialTransporter) Send(ctx context.Context, aduRequest []byte) (ad if err != nil { if mb.shouldRecover(err) { - if err = mb.reconnect(ctx, err, linkRecoveryDeadline); err != nil { + if err = mb.reconnect(ctx, err); err != nil { return } continue diff --git a/serial.go b/serial.go index f9199c4..a8d1c0a 100644 --- a/serial.go +++ b/serial.go @@ -19,8 +19,6 @@ const ( // Default timeout serialTimeout = 5 * time.Second serialIdleTimeout = 60 * time.Second - // Retry interval while spending the link recovery budget on reconnects. - serialReconnectRetryInterval = 10 * time.Millisecond ) // serialPort has configuration and I/O controller. @@ -33,17 +31,22 @@ type serialPort struct { IdleTimeout time.Duration // Silent period after successful connection ConnectDelay time.Duration - // Recovery timeout if the connection is lost + // Deprecated: Use LinkRecoveryBackoff.Timeout instead. LinkRecoveryTimeout time.Duration - // Interval between reconnect attempts while spending the link recovery budget. - // Zero or negative values fall back to the default retry interval. + // Deprecated: LinkRecoveryBackoff with FixedBackoff(). ReconnectRetryInterval time.Duration + // LinkRecoveryBackoff defines the unified retry strategy for link recovery. + // Controls retry intervals, timeout budget, and max attempts. + LinkRecoveryBackoff *BackoffStrategy + mu sync.Mutex // port is platform-dependent data structure for serial port. port io.ReadWriteCloser lastActivity time.Time closeTimer *time.Timer + // autoMigratedBackoff is a cached strategy created from deprecated fields + autoMigratedBackoff *BackoffStrategy } func (mb *serialPort) Connect(ctx context.Context) (err error) { @@ -102,9 +105,11 @@ func (mb *serialPort) shouldRecover(err error) bool { return errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) } -func (mb *serialPort) reconnect(ctx context.Context, err error, linkRecoveryDeadline time.Time) error { - if mb.LinkRecoveryTimeout == 0 || time.Until(linkRecoveryDeadline) < 0 { - return fmt.Errorf("modbus: link recovery timeout reached: %w", err) +func (mb *serialPort) reconnect(ctx context.Context, err error) error { + strategy := mb.getLinkRecoveryBackoff() + if strategy == nil { + // No recovery configured + return fmt.Errorf("modbus: no link recovery configured: %w", err) } mb.logf("modbus: connection reset, reconnecting") @@ -114,34 +119,65 @@ func (mb *serialPort) reconnect(ctx context.Context, err error, linkRecoveryDead mb.logf("modbus: error closing connection: %v", cerr) } - deadlineTimer := time.NewTimer(time.Until(linkRecoveryDeadline)) - defer deadlineTimer.Stop() - retryTicker := time.NewTicker(mb.reconnectRetryInterval()) - defer retryTicker.Stop() + start := time.Now() + attempt := 0 for { + elapsed := time.Since(start) + if !strategy.ShouldRetry(attempt, elapsed) { + return fmt.Errorf("modbus: link recovery exhausted: %w", recoveryErr) + } + if cerr := mb.connect(ctx); cerr == nil { return nil } else { recoveryErr = errors.Join(recoveryErr, cerr) - mb.logf("modbus: error reconnecting: %v", cerr) + mb.logf("modbus: reconnect attempt %d failed: %v", attempt, cerr) } + interval := strategy.Next(attempt) + attempt++ + select { case <-ctx.Done(): return ctx.Err() - case <-deadlineTimer.C: - return fmt.Errorf("modbus: link recovery timeout reached: %w", recoveryErr) - case <-retryTicker.C: + case <-time.After(interval): } } } -func (mb *serialPort) reconnectRetryInterval() time.Duration { - if mb.ReconnectRetryInterval > 0 { - return mb.ReconnectRetryInterval +func (mb *serialPort) getLinkRecoveryBackoff() *BackoffStrategy { + if mb.LinkRecoveryBackoff != nil { + return mb.LinkRecoveryBackoff } - return serialReconnectRetryInterval + + // Auto-migrate from deprecated fields + if mb.LinkRecoveryTimeout > 0 || mb.ReconnectRetryInterval > 0 { + if mb.autoMigratedBackoff == nil { + mb.logf("modbus: LinkRecoveryTimeout and ReconnectRetryInterval are deprecated, use LinkRecoveryBackoff instead") + interval := mb.ReconnectRetryInterval + if interval <= 0 { + interval = 100 * time.Millisecond + } + timeout := mb.LinkRecoveryTimeout + if timeout <= 0 { + timeout = 30 * time.Second // Default 30s timeout + } + // Use exponential backoff: 100ms→2s with configured timeout + mb.autoMigratedBackoff = &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: interval, + MaxInterval: 2 * time.Second, + Multiplier: 2.0, + Timeout: timeout, + MaxAttempts: 0, + } + } + return mb.autoMigratedBackoff + } + + // No recovery configured + return nil } func (mb *serialPort) startCloseTimer() { diff --git a/serial_test.go b/serial_test.go index 86cd010..c1f3c31 100644 --- a/serial_test.go +++ b/serial_test.go @@ -69,14 +69,14 @@ func TestSerialReconnect_UsesConfiguredRetryInterval(t *testing.T) { } s.Address = filepath.Join(t.TempDir(), "missing-serial") - err := s.reconnect(context.Background(), io.EOF, time.Now().Add(recoveryTimeout)) + err := s.reconnect(context.Background(), io.EOF) if err == nil { t.Fatal("expected reconnect to fail when the serial device is missing") } - if !strings.Contains(err.Error(), "link recovery timeout reached") { + if !strings.Contains(err.Error(), "link recovery exhausted") { t.Fatalf("expected link recovery timeout error, got %v", err) } - if count := strings.Count(logs.String(), "error reconnecting"); count != 1 { + if count := strings.Count(logs.String(), "reconnect attempt"); count != 1 { t.Fatalf("expected exactly one reconnect attempt before timeout, got %d logs: %q", count, logs.String()) } if !port.closed.Load() || s.port != nil { @@ -86,7 +86,7 @@ func TestSerialReconnect_UsesConfiguredRetryInterval(t *testing.T) { func TestSerialReconnect_DefaultRetryIntervalRetriesMultipleTimes(t *testing.T) { var logs bytes.Buffer - recoveryTimeout := 45 * time.Millisecond + recoveryTimeout := 250 * time.Millisecond // Longer timeout to accommodate 100ms default interval s := serialPort{ Logger: log.New(&logs, "", 0), @@ -95,14 +95,14 @@ func TestSerialReconnect_DefaultRetryIntervalRetriesMultipleTimes(t *testing.T) } s.Address = filepath.Join(t.TempDir(), "missing-serial") - err := s.reconnect(context.Background(), io.EOF, time.Now().Add(recoveryTimeout)) + err := s.reconnect(context.Background(), io.EOF) if err == nil { t.Fatal("expected reconnect to fail when the serial device is missing") } if !errors.Is(err, io.EOF) { t.Fatalf("expected reconnect to preserve the original EOF, got %v", err) } - if count := strings.Count(logs.String(), "error reconnecting"); count < 2 { + if count := strings.Count(logs.String(), "reconnect attempt"); count < 2 { t.Fatalf("expected default retry interval to attempt reconnect multiple times, got %d logs: %q", count, logs.String()) } if count := strings.Count(err.Error(), "could not open"); count < 2 { @@ -152,7 +152,7 @@ func TestSerialReconnectHotPlug_EventuallySucceedsWithinRecoveryWindow_PTY(t *te s.BaudRate = 19200 s.Timeout = 50 * time.Millisecond - err := s.reconnect(context.Background(), io.EOF, time.Now().Add(recoveryTimeout)) + err := s.reconnect(context.Background(), io.EOF) if err != nil { t.Fatalf("expected reconnect to succeed before timeout, got %v", err) } @@ -176,7 +176,188 @@ func TestSerialReconnectHotPlug_EventuallySucceedsWithinRecoveryWindow_PTY(t *te if !strings.Contains(logs.String(), "error closing connection") { t.Fatalf("expected close error to be logged, got %q", logs.String()) } - if count := strings.Count(logs.String(), "error reconnecting"); count < 1 { + if count := strings.Count(logs.String(), "reconnect attempt"); count < 1 { t.Fatalf("expected reconnect to log failed reopen attempts before success, got %d logs: %q", count, logs.String()) } } + +func TestSerialReconnect_WithExponentialBackoff(t *testing.T) { + var logs bytes.Buffer + recoveryTimeout := 100 * time.Millisecond + + s := serialPort{ + Logger: log.New(&logs, "", 0), + port: &nopCloser{ReadWriter: &bytes.Buffer{}}, + LinkRecoveryBackoff: &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 5 * time.Millisecond, + MaxInterval: 50 * time.Millisecond, + Multiplier: 2.0, + Timeout: recoveryTimeout, + MaxAttempts: 0, + }, + } + s.Address = filepath.Join(t.TempDir(), "missing-serial") + + start := time.Now() + err := s.reconnect(context.Background(), io.EOF) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected reconnect to fail when the serial device is missing") + } + if !strings.Contains(err.Error(), "link recovery exhausted") { + t.Fatalf("expected link recovery exhausted error, got %v", err) + } + + // With exponential backoff (5ms, 10ms, 20ms, 40ms, 50ms, 50ms...), + // we should get fewer attempts than with fixed 5ms interval + reconnectCount := strings.Count(logs.String(), "reconnect attempt") + if reconnectCount < 2 { + t.Fatalf("expected at least 2 reconnect attempts, got %d", reconnectCount) + } + if reconnectCount > 8 { + t.Fatalf("expected exponential backoff to limit attempts, got %d", reconnectCount) + } + + // Should respect the recovery timeout (with some tolerance for timing variance) + if elapsed > recoveryTimeout+30*time.Millisecond { + t.Fatalf("reconnect took too long: %v > %v", elapsed, recoveryTimeout) + } +} + +func TestSerialReconnect_WithLinearBackoff(t *testing.T) { + var logs bytes.Buffer + recoveryTimeout := 80 * time.Millisecond + + s := serialPort{ + Logger: log.New(&logs, "", 0), + port: &nopCloser{ReadWriter: &bytes.Buffer{}}, + LinkRecoveryBackoff: &BackoffStrategy{ + Type: BackoffLinear, + InitialInterval: 5 * time.Millisecond, + MaxInterval: 30 * time.Millisecond, + Multiplier: 1.0, + Timeout: recoveryTimeout, + MaxAttempts: 0, + }, + } + s.Address = filepath.Join(t.TempDir(), "missing-serial") + + err := s.reconnect(context.Background(), io.EOF) + if err == nil { + t.Fatal("expected reconnect to fail when the serial device is missing") + } + if !strings.Contains(err.Error(), "link recovery exhausted") { + t.Fatalf("expected link recovery timeout error, got %v", err) + } + + // With linear backoff (5ms, 10ms, 15ms, 20ms, 25ms, 30ms, 30ms...), + // should get a moderate number of attempts + reconnectCount := strings.Count(logs.String(), "reconnect attempt") + if reconnectCount < 2 { + t.Fatalf("expected at least 2 reconnect attempts, got %d", reconnectCount) + } +} + +func TestSerialReconnect_BackoffRespectsDeadline(t *testing.T) { + var logs bytes.Buffer + recoveryTimeout := 50 * time.Millisecond + + s := serialPort{ + Logger: log.New(&logs, "", 0), + port: &nopCloser{ReadWriter: &bytes.Buffer{}}, + LinkRecoveryBackoff: &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 5 * time.Second, + Multiplier: 2.0, + Timeout: recoveryTimeout, + MaxAttempts: 0, + }, + } + s.Address = filepath.Join(t.TempDir(), "missing-serial") + + start := time.Now() + err := s.reconnect(context.Background(), io.EOF) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected reconnect to fail") + } + if !strings.Contains(err.Error(), "link recovery exhausted") { + t.Fatalf("expected timeout error, got %v", err) + } + + // Must not exceed recovery timeout significantly (with tolerance for timing variance) + if elapsed > recoveryTimeout+30*time.Millisecond { + t.Fatalf("reconnect exceeded timeout: elapsed=%v timeout=%v", elapsed, recoveryTimeout) + } +} + +func TestSerialReconnect_BackwardCompatibility(t *testing.T) { + var logs bytes.Buffer + recoveryTimeout := 40 * time.Millisecond + + // Using deprecated ReconnectRetryInterval (no BackoffStrategy set) + s := serialPort{ + Logger: log.New(&logs, "", 0), + port: &nopCloser{ReadWriter: &bytes.Buffer{}}, + LinkRecoveryTimeout: recoveryTimeout, + ReconnectRetryInterval: 50 * time.Millisecond, // Will limit to 1 attempt + } + s.Address = filepath.Join(t.TempDir(), "missing-serial") + + err := s.reconnect(context.Background(), io.EOF) + if err == nil { + t.Fatal("expected reconnect to fail when the serial device is missing") + } + + // Should log deprecation warning + if !strings.Contains(logs.String(), "deprecated") { + t.Fatalf("expected deprecation warning, got %q", logs.String()) + } + + // Should still respect the old field and use 50ms interval + if count := strings.Count(logs.String(), "reconnect attempt"); count != 1 { + t.Fatalf("expected exactly one reconnect attempt with 50ms interval, got %d", count) + } +} + +func TestSerialReconnect_BackoffStrategyTakesPrecedence(t *testing.T) { + var logs bytes.Buffer + recoveryTimeout := 100 * time.Millisecond + + // Both fields set - LinkRecoveryBackoff should take precedence + s := serialPort{ + Logger: log.New(&logs, "", 0), + port: &nopCloser{ReadWriter: &bytes.Buffer{}}, + LinkRecoveryTimeout: recoveryTimeout, // This should be ignored + ReconnectRetryInterval: 50 * time.Millisecond, // This should be ignored + LinkRecoveryBackoff: &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 10 * time.Millisecond, + Timeout: recoveryTimeout, + MaxAttempts: 0, + }, + } + s.Address = filepath.Join(t.TempDir(), "missing-serial") + + err := s.reconnect(context.Background(), io.EOF) + if err == nil { + t.Fatal("expected reconnect to fail") + } + + // Should NOT log deprecation warning when LinkRecoveryBackoff is set + if strings.Contains(logs.String(), "deprecated") { + t.Fatalf("should not log deprecation when LinkRecoveryBackoff is set, got %q", logs.String()) + } + + // Should use LinkRecoveryBackoff (10ms fixed), not ReconnectRetryInterval (50ms) + // Expect multiple attempts with 10ms interval + reconnectCount := strings.Count(logs.String(), "reconnect attempt") + if reconnectCount < 5 { + t.Fatalf("expected multiple attempts with 10ms backoff, got %d", reconnectCount) + } +} diff --git a/tcpclient.go b/tcpclient.go index 0ff0c15..9476279 100644 --- a/tcpclient.go +++ b/tcpclient.go @@ -44,6 +44,8 @@ type TCPClientHandler struct { } // NewTCPClientHandler allocates a new TCPClientHandler with the given options. +// The handler uses exponential backoff (10ms-5s) with 30s timeout by default for link recovery. +// This is appropriate for TCP network links. For custom backoff, set LinkRecoveryBackoff explicitly. func NewTCPClientHandler(address string, options ...TCPClientHandlerOption) *TCPClientHandler { h := &TCPClientHandler{} for _, o := range options { @@ -55,6 +57,19 @@ func NewTCPClientHandler(address string, options ...TCPClientHandlerOption) *TCP if h.Dial == nil { h.Dial = defaultDialFunc(h.Timeout) } + // Default exponential backoff for TCP: 10ms initial, suitable for faster network recovery + h.LinkRecoveryBackoff = NewExponentialBackoff( + 10*time.Millisecond, // Initial interval (network-appropriate) + 5*time.Second, // Max interval + 30*time.Second, // Timeout + ) + // Default protocol recovery for transaction ID mismatches: 10ms with 100ms timeout + // Protocol recovery is just processing junk data, fail fast if it persists + h.ProtocolRecoveryBackoff = NewExponentialBackoff( + 10*time.Millisecond, // Initial interval + 50*time.Millisecond, // Max interval + 100*time.Millisecond, // Timeout (fail fast for junk data) + ) return h } @@ -167,12 +182,20 @@ type tcpTransporter struct { // If zero, disables the caching of TCP connections and only uses dialed // connections for a single Send. IdleTimeout time.Duration - // Recovery timeout if tcp communication misbehaves + // Deprecated: Use LinkRecoveryBackoff.Timeout instead. LinkRecoveryTimeout time.Duration - // Recovery timeout if the protocol is malformed, e.g. wrong transaction ID + // Deprecated: Use ProtocolRecoveryBackoff.Timeout instead. ProtocolRecoveryTimeout time.Duration // Silent period after successful connection ConnectDelay time.Duration + // LinkRecoveryBackoff defines the unified retry strategy for link recovery. + // Controls retry intervals, timeout budget, and max attempts for connection failures. + // If nil, uses deprecated LinkRecoveryTimeout if set, otherwise no recovery. + LinkRecoveryBackoff *BackoffStrategy + // ProtocolRecoveryBackoff defines the unified retry strategy for protocol recovery. + // Controls timeout budget and max attempts for transaction ID mismatches. + // If nil, uses deprecated ProtocolRecoveryTimeout if set, otherwise no recovery. + ProtocolRecoveryBackoff *BackoffStrategy // Transmission logger Logger Logger @@ -190,6 +213,10 @@ type tcpTransporter struct { lastSuccessfulTransactionID uint16 tlsConfig *tls.Config + + // Cached strategies created from deprecated timeout fields + autoMigratedLinkBackoff *BackoffStrategy + autoMigratedProtocolBackoff *BackoffStrategy } // helper value to signify what to do in Send @@ -211,10 +238,19 @@ func (mb *tcpTransporter) Send(ctx context.Context, aduRequest []byte) (aduRespo } var data [tcpMaxLength]byte - linkRecoveryDeadline := time.Now().Add(mb.LinkRecoveryTimeout) - protocolRecoveryDeadline := time.Now().Add(mb.ProtocolRecoveryTimeout) + linkStrategy := mb.getLinkRecoveryBackoff() + protoStrategy := mb.getProtocolRecoveryBackoff() + linkStart := time.Now() + linkAttempt := 0 for { + // Check link recovery budget if strategy is configured + if linkStrategy != nil && linkAttempt > 0 { + elapsed := time.Since(linkStart) + if !linkStrategy.ShouldRetry(linkAttempt, elapsed) { + return nil, fmt.Errorf("modbus: link recovery exhausted") + } + } // Establish a new connection if not connected if err = mb.connect(ctx); err != nil { err = fmt.Errorf("modbus: connect: %w", err) @@ -260,6 +296,22 @@ func (mb *tcpTransporter) Send(ctx context.Context, aduRequest []byte) (aduRespo } mb.lastAttemptedTransactionID = binary.BigEndian.Uint16(aduRequest) + + // Calculate deadlines from strategies + linkRecoveryDeadline := time.Time{} + if linkStrategy != nil && linkStrategy.Timeout > 0 { + linkRecoveryDeadline = linkStart.Add(linkStrategy.Timeout) + } else if mb.LinkRecoveryTimeout > 0 { + linkRecoveryDeadline = time.Now().Add(mb.LinkRecoveryTimeout) + } + + protocolRecoveryDeadline := time.Time{} + if protoStrategy != nil && protoStrategy.Timeout > 0 { + protocolRecoveryDeadline = time.Now().Add(protoStrategy.Timeout) + } else if mb.ProtocolRecoveryTimeout > 0 { + protocolRecoveryDeadline = time.Now().Add(mb.ProtocolRecoveryTimeout) + } + var res readResult aduResponse, res, err = mb.readResponse(aduRequest, data[:], linkRecoveryDeadline, protocolRecoveryDeadline) if err != nil { @@ -281,14 +333,29 @@ func (mb *tcpTransporter) Send(ctx context.Context, aduRequest []byte) (aduRespo err = fmt.Errorf("modbus: read response: %w", err) } else { mb.lastSuccessfulTransactionID = binary.BigEndian.Uint16(aduResponse) + linkAttempt = 0 // Reset on success } return case readResultRetry: mb.logf("modbus: retry reading response, because of %v", err) continue case readResultCloseRetry: - mb.logf("modbus: close connection and retry reading response, because of %v", err) + mb.logf("modbus: close connection and retry, attempt %d, because of %v", linkAttempt, err) mb.close() + + // Apply backoff delay if configured + if linkStrategy != nil { + interval := linkStrategy.Next(linkAttempt) + if interval > 0 { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(interval): + } + } + } + linkAttempt++ + select { case <-ctx.Done(): return nil, ctx.Err() @@ -306,13 +373,12 @@ func (mb *tcpTransporter) shouldRecover(err error) bool { return errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, syscall.ECONNRESET) } - func (mb *tcpTransporter) readResponse(aduRequest []byte, data []byte, recoveryDeadline time.Time, protocolDeadline time.Time) (aduResponse []byte, res readResult, err error) { // res is readResultDone by default, which either means we succeeded or err contains the fatal error for { if _, err = io.ReadFull(mb.conn, data[:tcpHeaderSize]); err != nil { // recovery disabled or deadline reached - report error - if mb.LinkRecoveryTimeout == 0 || time.Until(recoveryDeadline) < 0 { + if recoveryDeadline.IsZero() || time.Until(recoveryDeadline) < 0 { return } if mb.shouldRecover(err) { @@ -324,7 +390,7 @@ func (mb *tcpTransporter) readResponse(aduRequest []byte, data []byte, recoveryD aduResponse, err = mb.processResponse(data[:]) // this also does io if err != nil { // recovery disabled or deadline reached - report error - if mb.LinkRecoveryTimeout == 0 || time.Until(recoveryDeadline) < 0 { + if recoveryDeadline.IsZero() || time.Until(recoveryDeadline) < 0 { return } if mb.shouldRecover(err) { @@ -543,3 +609,51 @@ func (mb *tcpTransporter) closeIdle() { mb.close() } } + +func (mb *tcpTransporter) getLinkRecoveryBackoff() *BackoffStrategy { + if mb.LinkRecoveryBackoff != nil { + return mb.LinkRecoveryBackoff + } + + // Auto-migrate from deprecated LinkRecoveryTimeout + if mb.LinkRecoveryTimeout > 0 { + if mb.autoMigratedLinkBackoff == nil { + mb.logf("modbus: LinkRecoveryTimeout is deprecated, use LinkRecoveryBackoff instead") + mb.autoMigratedLinkBackoff = &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 0, // No delay by default + MaxInterval: 0, + Timeout: mb.LinkRecoveryTimeout, + MaxAttempts: 0, + } + } + return mb.autoMigratedLinkBackoff + } + + // No recovery configured by default + return nil +} + +func (mb *tcpTransporter) getProtocolRecoveryBackoff() *BackoffStrategy { + if mb.ProtocolRecoveryBackoff != nil { + return mb.ProtocolRecoveryBackoff + } + + // Auto-migrate from deprecated ProtocolRecoveryTimeout + if mb.ProtocolRecoveryTimeout > 0 { + if mb.autoMigratedProtocolBackoff == nil { + mb.logf("modbus: ProtocolRecoveryTimeout is deprecated, use ProtocolRecoveryBackoff instead") + mb.autoMigratedProtocolBackoff = &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 0, // No delay for protocol recovery + MaxInterval: 0, + Timeout: mb.ProtocolRecoveryTimeout, + MaxAttempts: 0, + } + } + return mb.autoMigratedProtocolBackoff + } + + // No protocol recovery by default + return nil +} diff --git a/tcpclient_test.go b/tcpclient_test.go index d5a9262..497cb12 100644 --- a/tcpclient_test.go +++ b/tcpclient_test.go @@ -639,6 +639,225 @@ func BenchmarkTCPEncoder(b *testing.B) { } } +func TestTCPReconnect_WithExponentialBackoff(t *testing.T) { + // Count reconnection attempts via dial calls + dialAttempts := 0 + handler := NewTCPClientHandler("test:502", WithDialer( + func(_ context.Context, _, _ string) (net.Conn, error) { + dialAttempts++ + // Create a connection that fails on read (triggers readResultCloseRetry) + _, cliConn := net.Pipe() + return &failReadConn{Conn: cliConn, readErr: io.EOF}, nil + }, + )) + handler.Timeout = 20 * time.Millisecond + handler.LinkRecoveryBackoff = &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 5 * time.Millisecond, + MaxInterval: 50 * time.Millisecond, + Multiplier: 2.0, + Timeout: 150 * time.Millisecond, + MaxAttempts: 0, + } + timeout := handler.LinkRecoveryBackoff.Timeout + + tr := &handler.tcpTransporter + req := []byte{0, 1, 0, 0, 0, 2, 0, 3} + + start := time.Now() + _, err := tr.Send(context.Background(), req) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected read errors to exhaust recovery timeout") + } + + // With exponential backoff (5ms, 10ms, 20ms, 40ms, 50ms...), + // should get multiple retry attempts but fewer than with no backoff + if dialAttempts < 2 { + t.Fatalf("expected multiple reconnect attempts with backoff, got %d", dialAttempts) + } + if dialAttempts > 10 { + t.Fatalf("expected exponential backoff to limit attempts, got %d", dialAttempts) + } + + // Verify it respects the timeout + if elapsed > timeout+50*time.Millisecond { + t.Fatalf("took too long: %v > %v", elapsed, timeout) + } + + t.Logf("Dial attempts: %d, elapsed: %v", dialAttempts, elapsed) +} + +func TestTCPReconnect_WithLinearBackoff(t *testing.T) { + dialAttempts := 0 + handler := NewTCPClientHandler("test:502", WithDialer( + func(_ context.Context, _, _ string) (net.Conn, error) { + dialAttempts++ + _, cliConn := net.Pipe() + return &failReadConn{Conn: cliConn, readErr: io.EOF}, nil + }, + )) + handler.Timeout = 20 * time.Millisecond + handler.LinkRecoveryBackoff = &BackoffStrategy{ + Type: BackoffLinear, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 30 * time.Millisecond, + Multiplier: 1.0, + Timeout: 120 * time.Millisecond, + MaxAttempts: 0, + } + timeout := handler.LinkRecoveryBackoff.Timeout + + tr := &handler.tcpTransporter + req := []byte{0, 1, 0, 0, 0, 2, 0, 3} + + start := time.Now() + _, err := tr.Send(context.Background(), req) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected connection to fail") + } + + // With linear backoff (10ms, 20ms, 30ms, 30ms...), should get moderate attempts + if dialAttempts < 2 { + t.Fatalf("expected multiple reconnect attempts, got %d", dialAttempts) + } + + // Should respect timeout + if elapsed > timeout+50*time.Millisecond { + t.Fatalf("took too long: %v", elapsed) + } + + t.Logf("Dial attempts: %d, elapsed: %v", dialAttempts, elapsed) +} + +func TestTCPReconnect_BackoffRespectsDeadline(t *testing.T) { + dialAttempts := 0 + handler := NewTCPClientHandler("test:502", WithDialer( + func(_ context.Context, _, _ string) (net.Conn, error) { + dialAttempts++ + _, cliConn := net.Pipe() + return &failReadConn{Conn: cliConn, readErr: io.EOF}, nil + }, + )) + handler.Timeout = 10 * time.Millisecond + handler.LinkRecoveryBackoff = &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 5 * time.Second, // Large max + Multiplier: 2.0, + Timeout: 60 * time.Millisecond, + MaxAttempts: 0, + } + timeout := handler.LinkRecoveryBackoff.Timeout + + tr := &handler.tcpTransporter + req := []byte{0, 1, 0, 0, 0, 2, 0, 3} + + start := time.Now() + _, err := tr.Send(context.Background(), req) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected connection to fail") + } + + // Must not exceed timeout significantly + if elapsed > timeout+50*time.Millisecond { + t.Fatalf("exceeded timeout: elapsed=%v timeout=%v", elapsed, timeout) + } + + t.Logf("Dial attempts: %d, elapsed: %v", dialAttempts, elapsed) +} + +func TestTCPReconnect_NoBackoffByDefault(t *testing.T) { + dialAttempts := 0 + handler := NewTCPClientHandler("test:502", WithDialer( + func(_ context.Context, _, _ string) (net.Conn, error) { + dialAttempts++ + _, cliConn := net.Pipe() + return &failReadConn{Conn: cliConn, readErr: io.EOF}, nil + }, + )) + handler.Timeout = 10 * time.Millisecond + handler.LinkRecoveryBackoff = &BackoffStrategy{ + Type: BackoffFixed, + InitialInterval: 0, // No delay + MaxInterval: 0, + Timeout: 60 * time.Millisecond, + MaxAttempts: 0, + } + // No delay between retries (interval = 0) + + tr := &handler.tcpTransporter + req := []byte{0, 1, 0, 0, 0, 2, 0, 3} + + start := time.Now() + _, err := tr.Send(context.Background(), req) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected connection to fail") + } + + // Without backoff, should get more attempts in the same time window + // compared to with backoff + if dialAttempts < 2 { + t.Fatalf("expected multiple fast reconnect attempts, got %d", dialAttempts) + } + + t.Logf("Dial attempts (no backoff): %d, elapsed: %v", dialAttempts, elapsed) +} + +func TestTCPReconnect_ResetAttemptOnSuccess(t *testing.T) { + attemptCount := 0 + srvConn, cliConn := net.Pipe() + t.Cleanup(func() { + srvConn.Close() + cliConn.Close() + }) + + handler := NewTCPClientHandler("test:502", WithDialer( + func(_ context.Context, _, _ string) (net.Conn, error) { + attemptCount++ + return cliConn, nil + }, + )) + handler.Timeout = 100 * time.Millisecond + handler.LinkRecoveryBackoff = &BackoffStrategy{ + Type: BackoffExponential, + InitialInterval: 10 * time.Millisecond, + MaxInterval: 100 * time.Millisecond, + Multiplier: 2.0, + Timeout: 200 * time.Millisecond, + MaxAttempts: 0, + } + + tr := &handler.tcpTransporter + + // Mock server response + go func() { + buf := make([]byte, 260) + _, _ = srvConn.Read(buf) + // Send valid response + response := []byte{0, 1, 0, 0, 0, 3, 0, 3, 0} + _, _ = srvConn.Write(response) + }() + + req := []byte{0, 1, 0, 0, 0, 3, 0, 3, 0} + _, err := tr.Send(context.Background(), req) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + + // Should only dial once for successful connection + if attemptCount != 1 { + t.Fatalf("expected 1 dial for successful connection, got %d", attemptCount) + } +} + func BenchmarkTCPDecoder(b *testing.B) { decoder := tcpPackager{ SlaveID: 10,