xgress close tweaks. Fixes #772

plorenz · plorenz · commit 818662bdaf3e · 2025-07-23T11:01:37.000-04:00
1. Don't try to re-transmit if the payload buffer is closed. Current we're only check if the whole xgress is closed
2. If we're unrouted, close the payload buffer immediately, rather than gracefully, as the data in the payload buffer nowhere to send any data
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,15 @@
+# Release notes 1.2.1
+
+## Issues Fixed and Dependency Updates
+
+* github.com/openziti/sdk-golang: [v1.2.0 -> v1.2.1](https://github.com/openziti/sdk-golang/compare/v1.2.0...v1.2.1)
+    * [Issue #772](https://github.com/openziti/sdk-golang/issues/772) - xgress close tweaks
+    * [Issue #769](https://github.com/openziti/sdk-golang/issues/769) - Require sdk flow control when using more than one default connection
+
+* github.com/openziti/channel/v4: [v4.2.16 -> v4.2.18](https://github.com/openziti/channel/compare/v4.2.16...v4.2.18)
+    * [Issue #201](https://github.com/openziti/channel/issues/201) - SendAndWait methods should return an error if the channel closes instead of blocking
+    * [Issue #199](https://github.com/openziti/channel/issues/199) - Reject multi-underlay connections that are the first connection for a channel, but aren't marked as such.
+
 # Release notes 1.2.0
 
 ## What's New
diff --git a/example/go.mod b/example/go.mod
@@ -81,7 +81,7 @@ require (
 	github.com/muhlemmer/gu v0.3.1 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
-	github.com/openziti/channel/v4 v4.2.16 // indirect
+	github.com/openziti/channel/v4 v4.2.18 // indirect
 	github.com/openziti/edge-api v0.26.46 // indirect
 	github.com/openziti/identity v1.0.109 // indirect
 	github.com/openziti/metrics v1.4.2 // indirect
diff --git a/example/go.sum b/example/go.sum
@@ -364,8 +364,8 @@ github.com/onsi/gomega v1.13.0 h1:7lLHu94wT9Ij0o6EWWclhu0aOh32VxhkwEJvzuWPeak=
 github.com/onsi/gomega v1.13.0/go.mod h1:lRk9szgn8TxENtWd0Tp4c3wjlRfMTMH27I+3Je41yGY=
 github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
 github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
-github.com/openziti/channel/v4 v4.2.16 h1:xOV9gkXBN8d4X4FFeguELTZtnpeRoApu09PAMTsDQZg=
-github.com/openziti/channel/v4 v4.2.16/go.mod h1:v8AOCXLqE++BHGmDK7Ut8Uc2nCGyOu9hjkJNoWgpcyU=
+github.com/openziti/channel/v4 v4.2.18 h1:RuyB26/p+3L3BCtTB7tUH3O0KwfkRXeeVc2fEN9WU1c=
+github.com/openziti/channel/v4 v4.2.18/go.mod h1:v8AOCXLqE++BHGmDK7Ut8Uc2nCGyOu9hjkJNoWgpcyU=
 github.com/openziti/edge-api v0.26.46 h1:cFOGXTIMXmxTgcM/bGd50OqZES/yT37MriojGJxozzo=
 github.com/openziti/edge-api v0.26.46/go.mod h1:sYHVpm26Jr1u7VooNJzTb2b2nGSlmCHMnbGC8XfWSng=
 github.com/openziti/foundation/v2 v2.0.70 h1:1g4jIl7nIdqW8oCgRcwlbNvK8dwk4mPyrwjjtjH1tp8=
diff --git a/example/influxdb-client-go/go.mod b/example/influxdb-client-go/go.mod
@@ -94,7 +94,7 @@ require (
 	github.com/muhlemmer/gu v0.3.1 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
-	github.com/openziti/channel/v4 v4.2.16 // indirect
+	github.com/openziti/channel/v4 v4.2.18 // indirect
 	github.com/openziti/edge-api v0.26.46 // indirect
 	github.com/openziti/foundation/v2 v2.0.70 // indirect
 	github.com/openziti/identity v1.0.109 // indirect
diff --git a/example/influxdb-client-go/go.sum b/example/influxdb-client-go/go.sum
@@ -418,8 +418,8 @@ github.com/onsi/gomega v1.13.0 h1:7lLHu94wT9Ij0o6EWWclhu0aOh32VxhkwEJvzuWPeak=
 github.com/onsi/gomega v1.13.0/go.mod h1:lRk9szgn8TxENtWd0Tp4c3wjlRfMTMH27I+3Je41yGY=
 github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
 github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
-github.com/openziti/channel/v4 v4.2.16 h1:xOV9gkXBN8d4X4FFeguELTZtnpeRoApu09PAMTsDQZg=
-github.com/openziti/channel/v4 v4.2.16/go.mod h1:v8AOCXLqE++BHGmDK7Ut8Uc2nCGyOu9hjkJNoWgpcyU=
+github.com/openziti/channel/v4 v4.2.18 h1:RuyB26/p+3L3BCtTB7tUH3O0KwfkRXeeVc2fEN9WU1c=
+github.com/openziti/channel/v4 v4.2.18/go.mod h1:v8AOCXLqE++BHGmDK7Ut8Uc2nCGyOu9hjkJNoWgpcyU=
 github.com/openziti/edge-api v0.26.46 h1:cFOGXTIMXmxTgcM/bGd50OqZES/yT37MriojGJxozzo=
 github.com/openziti/edge-api v0.26.46/go.mod h1:sYHVpm26Jr1u7VooNJzTb2b2nGSlmCHMnbGC8XfWSng=
 github.com/openziti/foundation/v2 v2.0.70 h1:1g4jIl7nIdqW8oCgRcwlbNvK8dwk4mPyrwjjtjH1tp8=
diff --git a/go.mod b/go.mod
@@ -18,7 +18,7 @@ require (
 	github.com/michaelquigley/pfxlog v0.6.10
 	github.com/mitchellh/go-ps v1.0.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openziti/channel/v4 v4.2.16
+	github.com/openziti/channel/v4 v4.2.18
 	github.com/openziti/edge-api v0.26.46
 	github.com/openziti/foundation/v2 v2.0.70
 	github.com/openziti/identity v1.0.109
diff --git a/go.sum b/go.sum
@@ -303,8 +303,8 @@ github.com/onsi/gomega v1.13.0 h1:7lLHu94wT9Ij0o6EWWclhu0aOh32VxhkwEJvzuWPeak=
 github.com/onsi/gomega v1.13.0/go.mod h1:lRk9szgn8TxENtWd0Tp4c3wjlRfMTMH27I+3Je41yGY=
 github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
 github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
-github.com/openziti/channel/v4 v4.2.16 h1:xOV9gkXBN8d4X4FFeguELTZtnpeRoApu09PAMTsDQZg=
-github.com/openziti/channel/v4 v4.2.16/go.mod h1:v8AOCXLqE++BHGmDK7Ut8Uc2nCGyOu9hjkJNoWgpcyU=
+github.com/openziti/channel/v4 v4.2.18 h1:RuyB26/p+3L3BCtTB7tUH3O0KwfkRXeeVc2fEN9WU1c=
+github.com/openziti/channel/v4 v4.2.18/go.mod h1:v8AOCXLqE++BHGmDK7Ut8Uc2nCGyOu9hjkJNoWgpcyU=
 github.com/openziti/edge-api v0.26.46 h1:cFOGXTIMXmxTgcM/bGd50OqZES/yT37MriojGJxozzo=
 github.com/openziti/edge-api v0.26.46/go.mod h1:sYHVpm26Jr1u7VooNJzTb2b2nGSlmCHMnbGC8XfWSng=
 github.com/openziti/foundation/v2 v2.0.70 h1:1g4jIl7nIdqW8oCgRcwlbNvK8dwk4mPyrwjjtjH1tp8=
diff --git a/xgress/retransmitter.go b/xgress/retransmitter.go
@@ -145,8 +145,8 @@ func (self *Retransmitter) retransmitSender() {
 			if !retransmit.isAcked() {
 				retransmit.payload.MarkAsRetransmit()
 				if err := retransmit.x.dataPlane.RetransmitPayload(retransmit.x.address, retransmit.payload); err != nil {
-					// if xgress is closed, don't log the error. We still want to try retransmitting in case we're re-sending end of circuit
-					if !retransmit.x.Closed() {
+					// if xgress payload buffer is closed, don't log the error
+					if !retransmit.x.payloadBuffer.IsClosed() {
 						logger.WithError(err).Errorf("unexpected error while retransmitting payload from [@/%v]", retransmit.x.address)
 						self.retransmissionFailures.Mark(1)
 						self.faultReporter.ReportForwardingFault(retransmit.payload.CircuitId, retransmit.x.ctrlId)
diff --git a/xgress/xgress.go b/xgress/xgress.go
@@ -304,12 +304,11 @@ func (self *Xgress) CloseRxTimeout() {
 }
 
 func (self *Xgress) Unrouted() {
-	// if we're unrouted no more data is inbound
+	// if we're unrouted no more data is inbound. this still allows already queued data to flow to the client
 	self.CloseXgToClient()
 
-	// When we're unrouted, if 'end of circuit' hasn't already arrived, give incoming/queued data
-	// a chance to outflow before closing. We're unrouted so no point in sending EOF
-	self.payloadBuffer.CloseWhenEmpty()
+	// If we're unrouted, we can't send any more data, so close the payload buffer
+	self.payloadBuffer.Close()
 }
 
 func (self *Xgress) CloseXgToClient() {
diff --git a/ziti/edge/network/xg_adapter.go b/ziti/edge/network/xg_adapter.go
@@ -44,6 +44,10 @@ func (self *XgAdapter) RetransmitPayload(srcAddr xgress.Address, payload *xgress
 	msg := payload.Marshall()
 	sent, err := self.conn.MsgChannel.GetDefaultSender().TrySend(msg)
 	if err != nil {
+		// if the channel is closed, close the xgress
+		if self.conn.MsgChannel.GetChannel().IsClosed() {
+			self.xg.Close()
+		}
 		return err
 	}
 
diff --git a/ziti/sdkinfo/build_info.go b/ziti/sdkinfo/build_info.go
diff --git a/ziti/ziti.go b/ziti/ziti.go
@@ -1441,6 +1441,7 @@ func (context *ContextImpl) connectEdgeRouter(routerName, ingressUrl string) *ed
 	if context.maxControlConnections > 0 || context.maxDefaultConnections > 1 {
 		headers.PutBoolHeader(channel.IsGroupedHeader, true)
 		headers.PutStringHeader(channel.TypeHeader, edge.ChannelTypeDefault)
+		headers.PutBoolHeader(channel.IsFirstGroupConnection, true)
 	}
 	underlay, err := dialer.CreateWithHeaders(options.ConnectTimeout, headers)
 	if err != nil {

Original file line number	Diff line number	Diff line change
`@@ -304,12 +304,11 @@ func (self *Xgress) CloseRxTimeout() {`
`304`	`304`	`}`
`305`	`305`
`306`	`306`	`func (self *Xgress) Unrouted() {`
`307`		`- // if we're unrouted no more data is inbound`
	`307`	`+ // if we're unrouted no more data is inbound. this still allows already queued data to flow to the client`
`308`	`308`	`self.CloseXgToClient()`
`309`	`309`
`310`		`- // When we're unrouted, if 'end of circuit' hasn't already arrived, give incoming/queued data`
`311`		`- // a chance to outflow before closing. We're unrouted so no point in sending EOF`
`312`		`- self.payloadBuffer.CloseWhenEmpty()`
	`310`	`+ // If we're unrouted, we can't send any more data, so close the payload buffer`
	`311`	`+ self.payloadBuffer.Close()`
`313`	`312`	`}`
`314`	`313`
`315`	`314`	`func (self *Xgress) CloseXgToClient() {`