contract,transport: reduce log spam during backend outages

transfer.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"math"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	// "runtime/debug"
@@ -23,6 +24,26 @@ import (
 	"github.com/urnetwork/connect/protocol"
 )
 
+var (
+	lastDropErrLogNano     atomic.Int64
+	suppressedDropErrCount atomic.Int64
+)
+
+func shouldLogDropErr() (bool, int64) {
+	now := time.Now().UnixNano()
+	last := lastDropErrLogNano.Load()
+	if now-last < int64(time.Minute) {
+		suppressedDropErrCount.Add(1)
+		return false, 0
+	}
+	if !lastDropErrLogNano.CompareAndSwap(last, now) {
+		suppressedDropErrCount.Add(1)
+		return false, 0
+	}
+	suppressed := suppressedDropErrCount.Swap(0)
+	return true, suppressed
+}
+
 /*
 Sends frames to destinations with properties:
 - as long the sending client is active, frames are eventually delivered up to timeout
@@ -3631,7 +3652,13 @@ func (self *ReceiveSequence) Run() {
 			} else {
 				err := c()
 				if err != nil {
-					self.log.Infof("[r]drop = %s", err)
+					if ok, suppressed := shouldLogDropErr(); ok {
+						if suppressed > 0 {
+							self.log.Infof("[r]drop = %s (%d suppressed)", err, suppressed)
+						} else {
+							self.log.Infof("[r]drop = %s", err)
+						}
+					}
 				}
 			}
 		}

transfer_contract_manager.go

@@ -3,6 +3,7 @@ package connect
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	// "errors"
@@ -22,6 +23,24 @@ import (
 	"github.com/urnetwork/connect/protocol"
 )
 
+var lastOobErrLogNano atomic.Int64
+var suppressedOobErrCount atomic.Int64
+
+func shouldLogOobErr() (bool, int64) {
+	now := time.Now().UnixNano()
+	last := lastOobErrLogNano.Load()
+	if now-last < int64(time.Minute) {
+		suppressedOobErrCount.Add(1)
+		return false, 0
+	}
+	if !lastOobErrLogNano.CompareAndSwap(last, now) {
+		suppressedOobErrCount.Add(1)
+		return false, 0
+	}
+	suppressed := suppressedOobErrCount.Swap(0)
+	return true, suppressed
+}
+
 // manage contracts which are embedded into each transfer sequence
 
 type ContractKey struct {
@@ -206,6 +225,8 @@ func DefaultContractManagerSettingsWithBufferSize(bufferSize int) *ContractManag
 
 		ContractQueueExpireTimeout: 120 * time.Second,
 
+		CreateContractOobErrorBackoff: time.Minute,
+
 		ProtocolVersion: DefaultProtocolVersion,
 
 		// TODO remove
@@ -259,6 +280,10 @@ type ContractManagerSettings struct {
 	// window (5 minutes). <= 0 disables expiry.
 	ContractQueueExpireTimeout time.Duration
 
+	// back off create-contract OOB API calls after an OOB error to avoid
+	// repeatedly hitting the API while it is timing out or unavailable.
+	CreateContractOobErrorBackoff time.Duration
+
 	ProtocolVersion int
 
 	// TODO remove
@@ -299,12 +324,38 @@ type ContractManager struct {
 
 	controlSyncProvide    *ControlSync
 	controlSyncProvideOob *ControlSyncOob
+
+	createContractOobErrorBackoffUntil time.Time
 }
 
 func NewContractManagerWithDefaults(ctx context.Context, client *Client) *ContractManager {
 	return NewContractManager(ctx, client, DefaultContractManagerSettings())
 }
 
+func (self *ContractManager) createContractOobErrorBackoffActive() bool {
+	if self.settings.CreateContractOobErrorBackoff <= 0 {
+		return false
+	}
+
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+
+	return time.Now().Before(self.createContractOobErrorBackoffUntil)
+}
+
+func (self *ContractManager) markCreateContractOobError() {
+	if self.settings.CreateContractOobErrorBackoff <= 0 {
+		return
+	}
+
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+
+	if !time.Now().Before(self.createContractOobErrorBackoffUntil) {
+		self.createContractOobErrorBackoffUntil = time.Now().Add(self.settings.CreateContractOobErrorBackoff)
+	}
+}
+
 func NewContractManager(
 	ctx context.Context,
 	client *Client,
@@ -1055,6 +1106,10 @@ func (self *ContractManager) addContract(contractKey ContractKey, contract *prot
 }
 
 func (self *ContractManager) CreateContract(contractKey ContractKey, contractSeqIndex uint64, minByteCount ByteCount) {
+	if self.createContractOobErrorBackoffActive() {
+		return
+	}
+
 	// look at destinationContracts and last contract to get previous contract id
 	contractQueue := self.openContractQueue(contractKey)
 	defer self.closeContractQueue(contractKey)
@@ -1092,7 +1147,14 @@ func (self *ContractManager) CreateContract(contractKey ContractKey, contractSeq
 				case <-self.client.Done():
 					// no need to log warnings when the client closes
 				default:
-					self.client.log.Infof("[contract]oob err = %s\n", err)
+					self.markCreateContractOobError()
+					if ok, suppressed := shouldLogOobErr(); ok {
+						if suppressed > 0 {
+							self.client.log.Infof("[contract]oob err = %s; backing off create contract OOB requests for %s (%d suppressed)\n", err, self.settings.CreateContractOobErrorBackoff, suppressed)
+						} else {
+							self.client.log.Infof("[contract]oob err = %s; backing off create contract OOB requests for %s\n", err, self.settings.CreateContractOobErrorBackoff)
+						}
+					}
 				}
 			}
 		},

transport.go

@@ -186,6 +186,7 @@ type PlatformTransport struct {
 	availableModes map[TransportMode]bool
 	targetMode     TransportMode
 	mode           TransportMode
+
 }
 
 func NewPlatformTransportWithDefaults(
@@ -390,6 +391,56 @@ func isBetterMode(current TransportMode, other TransportMode) bool {
 	return transportModePreferences[current] < transportModePreferences[other]
 }
 
+// lastAuthErrLogNano and suppressedAuthErrCount are package-level atomics shared
+// across all PlatformTransport instances, rate-limiting [t]auth error log lines to
+// at most once per minute and tracking how many were suppressed in the interval.
+var lastAuthErrLogNano atomic.Int64
+var suppressedAuthErrCount atomic.Int64
+
+// lastWriteErrLogNano and suppressedWriteErrCount rate-limit [ts] write-error
+// log lines. A socket that is broken but still open keeps accepting writes that
+// immediately fail, so the writer goroutine retries in a tight loop and every
+// failed write logs a line. During an outage this floods the log just like the
+// auth path, so the same once-per-minute throttle is applied here.
+var lastWriteErrLogNano atomic.Int64
+var suppressedWriteErrCount atomic.Int64
+
+// shouldLogAuthErr returns (true, suppressedCount) if a log line should be emitted,
+// resetting the suppressed counter. Returns (false, 0) if the error is suppressed.
+func shouldLogAuthErr() (bool, int64) {
+	now := time.Now().UnixNano()
+	last := lastAuthErrLogNano.Load()
+	if now-last < int64(time.Minute) {
+		suppressedAuthErrCount.Add(1)
+		return false, 0
+	}
+	if !lastAuthErrLogNano.CompareAndSwap(last, now) {
+		suppressedAuthErrCount.Add(1)
+		return false, 0
+	}
+	suppressed := suppressedAuthErrCount.Swap(0)
+	return true, suppressed
+}
+
+// shouldLogWriteErr returns (true, suppressedCount) if a [ts] write-error line
+// should be emitted, resetting the suppressed counter. Returns (false, 0) if
+// suppressed. A broken-but-open socket retries writes in a tight loop; this
+// throttles that flood to one line per minute with a suppressed count.
+func shouldLogWriteErr() (bool, int64) {
+	now := time.Now().UnixNano()
+	last := lastWriteErrLogNano.Load()
+	if now-last < int64(time.Minute) {
+		suppressedWriteErrCount.Add(1)
+		return false, 0
+	}
+	if !lastWriteErrLogNano.CompareAndSwap(last, now) {
+		suppressedWriteErrCount.Add(1)
+		return false, 0
+	}
+	suppressed := suppressedWriteErrCount.Swap(0)
+	return true, suppressed
+}
+
 func (self *PlatformTransport) runH1(initialTimeout time.Duration) {
 	// connect and update route manager for this transport
 	defer self.cancel()
@@ -404,6 +455,7 @@ func (self *PlatformTransport) runH1(initialTimeout time.Duration) {
 		}
 	}
 
+	authErrLogged := false
 	for {
 		// wait until we are back in h1 or worse
 		func() {
@@ -494,14 +546,28 @@ func (self *PlatformTransport) runH1(initialTimeout time.Duration) {
 			ws, err = connect()
 		}
 		if err != nil {
-			self.log.Infof("[t]auth error %s = %s\n", clientId, err)
+			if !authErrLogged {
+				if ok, suppressed := shouldLogAuthErr(); ok {
+					if suppressed > 0 {
+						self.log.Infof("[t]auth error %s = %s (%d suppressed)\n", clientId, err, suppressed)
+					} else {
+						self.log.Infof("[t]auth error %s = %s\n", clientId, err)
+					}
+					authErrLogged = true
+				} else {
+					self.log.V(1).Infof("[t]auth error %s = %s\n", clientId, err)
+				}
+			} else {
+				self.log.V(1).Infof("[t]auth error %s = %s\n", clientId, err)
+			}
 			select {
 			case <-self.ctx.Done():
 				return
 			case <-reconnect.After():
 				continue
 			}
 		}
+		authErrLogged = false
 
 		c := func() {
 			defer ws.Close()
@@ -643,7 +709,13 @@ func (self *PlatformTransport) runH1(initialTimeout time.Duration) {
 					MessagePoolReturn(message)
 					if err != nil {
 						// note that for websocket a dealine timeout cannot be recovered
-						self.log.Infof("[ts]%s-> error = %s\n", clientId, err)
+						if ok, suppressed := shouldLogWriteErr(); ok {
+							if suppressed > 0 {
+								self.log.Infof("[ts]%s-> error = %s (%d suppressed)\n", clientId, err, suppressed)
+							} else {
+								self.log.Infof("[ts]%s-> error = %s\n", clientId, err)
+							}
+						}
 						return err
 					}
 					self.log.V(2).Infof("[ts]%s->\n", clientId)
@@ -900,6 +972,7 @@ func (self *PlatformTransport) runH3(ptMode TransportMode, initialTimeout time.D
 		}
 	}
 
+	authErrLogged := false
 	for {
 		// wait until we are back in the specific pt mode or auto mode
 		func() {
@@ -1079,14 +1152,28 @@ func (self *PlatformTransport) runH3(ptMode TransportMode, initialTimeout time.D
 			connStream, err = connect()
 		}
 		if err != nil {
-			self.log.Infof("[t]auth error %s = %s\n", clientId, err)
+			if !authErrLogged {
+				if ok, suppressed := shouldLogAuthErr(); ok {
+					if suppressed > 0 {
+						self.log.Infof("[t]auth error %s = %s (%d suppressed)\n", clientId, err, suppressed)
+					} else {
+						self.log.Infof("[t]auth error %s = %s\n", clientId, err)
+					}
+					authErrLogged = true
+				} else {
+					self.log.V(1).Infof("[t]auth error %s = %s\n", clientId, err)
+				}
+			} else {
+				self.log.V(1).Infof("[t]auth error %s = %s\n", clientId, err)
+			}
 			select {
 			case <-self.ctx.Done():
 				return
 			case <-reconnect.After():
 				continue
 			}
 		}
+		authErrLogged = false
 		conn := connStream.conn
 		stream := connStream.stream