Merge pull request #773 from openziti/close-tweaks

plorenz · web-flow · commit 8c177bb0788b · 2025-07-25T10:33:23.000-04:00
xgress close tweaks. Fixes #772
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/conn.go b/ziti/edge/network/conn.go
@@ -108,11 +108,12 @@ func (conn *edgeConn) CloseWrite() error {
 		headers := channel.Headers{}
 		headers.PutUint32Header(edge.FlagsHeader, edge.FIN)
 		_, err := conn.WriteTraced(nil, nil, headers)
-		return err
-	}
 
-	if conn.xgCircuit != nil {
-		conn.xgCircuit.xg.CloseRxTimeout()
+		if conn.xgCircuit != nil {
+			conn.xgCircuit.xg.CloseRxTimeout()
+		}
+
+		return err
 	}
 
 	return nil
diff --git a/ziti/edge/network/conn_test.go b/ziti/edge/network/conn_test.go
@@ -185,6 +185,10 @@ func TestReadMultipart(t *testing.T) {
 type NoopTestChannel struct {
 }
 
+func (ch *NoopTestChannel) CloseNotify() <-chan struct{} {
+	panic("implement me")
+}
+
 func (ch *NoopTestChannel) GetUnderlays() []channel.Underlay {
 	panic("implement me")
 }
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() {`