Fix ListenOptions UseSdkFlowControl and update changelog

plorenz · plorenz · commit 091a7926e8f3 · 2025-04-29T20:25:59.000-04:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,14 +14,18 @@ a slow circuit can get stalled at the router because of flow control back-pressu
 This then back-pressures all circuits from the SDK to that router. 
 
 By moving the flow-control to the SDK, a slow circuit will not negatively impact
-other circuits to the same router. This is currently enabled in the `DialOptions`.
+other circuits to the same router. This is currently enabled in the `DialOptions`
+and `ListenOptions` for the dial and hosting sides respectively.
 
 ```
 t := true
 dialOptions := &ziti.DialOptions{
     ConnectTimeout: wf.ConnectTimeout,
     SdkFlowControl: &t,
 }
+
+listenOptions := ziti.DefaultListenOptions()
+listenOptions.SdkFlowControl = &t
 ```
 
 As this is an experimental feature, the configuration may change or be removed
diff --git a/example/go.mod b/example/go.mod
@@ -83,7 +83,7 @@ require (
 	github.com/openziti/channel/v4 v4.0.6 // indirect
 	github.com/openziti/edge-api v0.26.42 // indirect
 	github.com/openziti/identity v1.0.101 // indirect
-	github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83 // indirect
+	github.com/openziti/metrics v1.4.1 // indirect
 	github.com/openziti/secretstream v0.1.32 // indirect
 	github.com/orcaman/concurrent-map/v2 v2.0.1 // indirect
 	github.com/parallaxsecond/parsec-client-go v0.0.0-20221025095442-f0a77d263cf9 // indirect
diff --git a/example/go.sum b/example/go.sum
@@ -366,8 +366,8 @@ github.com/openziti/foundation/v2 v2.0.59 h1:PJwrcTq62x+cONBeKMlnsuphsTlOvTz8j8p
 github.com/openziti/foundation/v2 v2.0.59/go.mod h1:76gmsdIBHvv4O3I0TuFBfO58Rv7YN8FA0ojwYz27ZxE=
 github.com/openziti/identity v1.0.101 h1:XujPT6eCv3Gqsodh9846EoWfQZwDAFrjPLxm1cipveY=
 github.com/openziti/identity v1.0.101/go.mod h1:OZLcSf2wGqAa7/8SmbwLQxv0Gsv4i4pxTk83WPHIr+o=
-github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83 h1:XJfBC7Yd6bQCUsGaCXqoOyBNTtdohGepxTveFv7Yn7E=
-github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg=
+github.com/openziti/metrics v1.4.1 h1:JMnfmW8liTzmqJpd7HDPCSvDDFg1UrE0WoLhW9XdIG8=
+github.com/openziti/metrics v1.4.1/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg=
 github.com/openziti/runzmd v1.0.33 h1:tOyjRoUuVXIo1z1pNU32jALWkMmhzsSaDrhLtuOn3Ts=
 github.com/openziti/runzmd v1.0.33/go.mod h1:8c/uvZR/XWXQNllTq6LuTpfKL2DTNxfI2X2wYhgRwik=
 github.com/openziti/secretstream v0.1.32 h1:89/ZVcwIQjdVmWDfVRfMEChJJXTLXJ59AYBw5j646M4=
diff --git a/example/influxdb-client-go/go.mod b/example/influxdb-client-go/go.mod
@@ -98,7 +98,7 @@ require (
 	github.com/openziti/edge-api v0.26.42 // indirect
 	github.com/openziti/foundation/v2 v2.0.59 // indirect
 	github.com/openziti/identity v1.0.101 // indirect
-	github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83 // indirect
+	github.com/openziti/metrics v1.4.1 // indirect
 	github.com/openziti/secretstream v0.1.32 // indirect
 	github.com/openziti/transport/v2 v2.0.168 // indirect
 	github.com/orcaman/concurrent-map/v2 v2.0.1 // indirect
diff --git a/example/influxdb-client-go/go.sum b/example/influxdb-client-go/go.sum
@@ -421,8 +421,8 @@ github.com/openziti/foundation/v2 v2.0.59 h1:PJwrcTq62x+cONBeKMlnsuphsTlOvTz8j8p
 github.com/openziti/foundation/v2 v2.0.59/go.mod h1:76gmsdIBHvv4O3I0TuFBfO58Rv7YN8FA0ojwYz27ZxE=
 github.com/openziti/identity v1.0.101 h1:XujPT6eCv3Gqsodh9846EoWfQZwDAFrjPLxm1cipveY=
 github.com/openziti/identity v1.0.101/go.mod h1:OZLcSf2wGqAa7/8SmbwLQxv0Gsv4i4pxTk83WPHIr+o=
-github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83 h1:XJfBC7Yd6bQCUsGaCXqoOyBNTtdohGepxTveFv7Yn7E=
-github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg=
+github.com/openziti/metrics v1.4.1 h1:JMnfmW8liTzmqJpd7HDPCSvDDFg1UrE0WoLhW9XdIG8=
+github.com/openziti/metrics v1.4.1/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg=
 github.com/openziti/secretstream v0.1.32 h1:89/ZVcwIQjdVmWDfVRfMEChJJXTLXJ59AYBw5j646M4=
 github.com/openziti/secretstream v0.1.32/go.mod h1:8YaIbjyMwBeKQ7eOYcoVPKHT10u+4OVPXpnZAeDzC6o=
 github.com/openziti/transport/v2 v2.0.168 h1:1Anf7X+4xmSKQ12GdPJFhoMZi04QxgD4MJu3agFc1R4=
diff --git a/go.mod b/go.mod
@@ -22,7 +22,7 @@ require (
 	github.com/openziti/edge-api v0.26.42
 	github.com/openziti/foundation/v2 v2.0.59
 	github.com/openziti/identity v1.0.101
-	github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83
+	github.com/openziti/metrics v1.4.1
 	github.com/openziti/secretstream v0.1.32
 	github.com/openziti/transport/v2 v2.0.168
 	github.com/orcaman/concurrent-map/v2 v2.0.1
diff --git a/go.sum b/go.sum
@@ -307,8 +307,8 @@ github.com/openziti/foundation/v2 v2.0.59 h1:PJwrcTq62x+cONBeKMlnsuphsTlOvTz8j8p
 github.com/openziti/foundation/v2 v2.0.59/go.mod h1:76gmsdIBHvv4O3I0TuFBfO58Rv7YN8FA0ojwYz27ZxE=
 github.com/openziti/identity v1.0.101 h1:XujPT6eCv3Gqsodh9846EoWfQZwDAFrjPLxm1cipveY=
 github.com/openziti/identity v1.0.101/go.mod h1:OZLcSf2wGqAa7/8SmbwLQxv0Gsv4i4pxTk83WPHIr+o=
-github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83 h1:XJfBC7Yd6bQCUsGaCXqoOyBNTtdohGepxTveFv7Yn7E=
-github.com/openziti/metrics v1.4.1-0.20250428140813-8f25360e7e83/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg=
+github.com/openziti/metrics v1.4.1 h1:JMnfmW8liTzmqJpd7HDPCSvDDFg1UrE0WoLhW9XdIG8=
+github.com/openziti/metrics v1.4.1/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg=
 github.com/openziti/secretstream v0.1.32 h1:89/ZVcwIQjdVmWDfVRfMEChJJXTLXJ59AYBw5j646M4=
 github.com/openziti/secretstream v0.1.32/go.mod h1:8YaIbjyMwBeKQ7eOYcoVPKHT10u+4OVPXpnZAeDzC6o=
 github.com/openziti/transport/v2 v2.0.168 h1:1Anf7X+4xmSKQ12GdPJFhoMZi04QxgD4MJu3agFc1R4=
diff --git a/xgress/link_send_buffer.go b/xgress/link_send_buffer.go
@@ -245,7 +245,9 @@ func (buffer *LinkSendBuffer) run() {
 		case ack := <-buffer.newlyReceivedAcks:
 			buffer.receiveAcknowledgement(ack)
 			buffer.retransmit()
-			buffer.checkForCloseOnEmpty()
+			if buffer.closeWhenEmpty.Load() && len(buffer.buffer) == 0 && !buffer.x.Closed() && buffer.x.IsEndOfCircuitSent() {
+				go buffer.x.Close()
+			}
 
 		case txPayload := <-buffered:
 			buffer.buffer[txPayload.payload.GetSequence()] = txPayload
@@ -257,7 +259,6 @@ func (buffer *LinkSendBuffer) run() {
 
 		case <-retransmitTicker.C:
 			buffer.retransmit()
-			buffer.checkForCloseOnEmpty()
 
 		case <-buffer.closeNotify:
 			buffer.close()
@@ -266,12 +267,6 @@ func (buffer *LinkSendBuffer) run() {
 	}
 }
 
-func (buffer *LinkSendBuffer) checkForCloseOnEmpty() {
-	if buffer.closeWhenEmpty.Load() && len(buffer.buffer) == 0 && !buffer.x.Closed() && buffer.x.IsEndOfCircuitSent() {
-		go buffer.x.Close()
-	}
-}
-
 func (buffer *LinkSendBuffer) close() {
 	if buffer.blockedByLocalWindow {
 		buffer.metrics().BufferUnblockedByLocalWindow()
diff --git a/ziti/ziti.go b/ziti/ziti.go
@@ -1263,9 +1263,7 @@ func (context *ContextImpl) listenSession(service *rest_model.ServiceDetail, opt
 		edgeListenOptions.MaxTerminators = 1
 	}
 
-	// TODO: uncomment
-	//edgeListenOptions.SdkFlowControl = options.SdkFlowControl != nil && *options.SdkFlowControl
-	edgeListenOptions.SdkFlowControl = true
+	edgeListenOptions.SdkFlowControl = options.SdkFlowControl != nil && *options.SdkFlowControl
 
 	if listenerMgr, err := newListenerManager(service, context, edgeListenOptions, options.WaitForNEstablishedListeners); err != nil {
 		return nil, err

Original file line number	Diff line number	Diff line change
`@@ -1263,9 +1263,7 @@ func (context ContextImpl) listenSession(service rest_model.ServiceDetail, opt`
`1263`	`1263`	`edgeListenOptions.MaxTerminators = 1`
`1264`	`1264`	`}`
`1265`	`1265`
`1266`		`- // TODO: uncomment`
`1267`		`- //edgeListenOptions.SdkFlowControl = options.SdkFlowControl != nil && *options.SdkFlowControl`
`1268`		`- edgeListenOptions.SdkFlowControl = true`
	`1266`	`+ edgeListenOptions.SdkFlowControl = options.SdkFlowControl != nil && *options.SdkFlowControl`
`1269`	`1267`
`1270`	`1268`	`if listenerMgr, err := newListenerManager(service, context, edgeListenOptions, options.WaitForNEstablishedListeners); err != nil {`
`1271`	`1269`	`return nil, err`