refactor(dgram): name proxy copy directions

commit: 8916627801043752233c4dbdde884dc3083ca1e6
author: brice · 3mo ago · 2026-07-08 11:20 UTC
parent: f1350fe
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +64 -13

M internal/adapters/dgram/proxy.go +31 -12


@@ -45,6 +45,22 @@ const (
45	45		proxyRetryDone
46	46		)
47	47
	48	+	type proxyCopier struct {
	49	+	ctx context.Context
	50	+	errCh chan<- proxyCopyErr
	51	+	retryEvents chan<- proxyRetryEvent
	52	+	retryBackoff time.Duration
	53	+	clk ports.Clock
	54	+	}
	55	+
	56	+	type proxyCopyDirection struct {
	57	+	src ports.Transport
	58	+	dst ports.Transport
	59	+	recvKind proxyCopyErrKind
	60	+	sendKind proxyCopyErrKind
	61	+	retryPendingFull bool
	62	+	}
	63	+
48	64		func (p ProxyRuntime) Run(ctx context.Context) error {
49	65		defer func() {
50	66		_ = p.Client.Close()

@@ -66,11 +82,14 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
66	82
67	83		errCh := make(chan proxyCopyErr, 2)
68	84		retryEvents := make(chan proxyRetryEvent, 2)
	85	+	copier := proxyCopier{ctx: ctx, errCh: errCh, retryEvents: retryEvents, retryBackoff: retryBackoff, clk: clk}
	86	+	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}
	87	+	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryPendingFull: true}
69	88		startClientCopy := func() {
70		-	go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryEvents, retryBackoff, clk, false)
	89	+	go copier.copyFrames(clientToDaemon)
71	90		}
72	91		startDaemonCopy := func() {
73		-	go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryEvents, retryBackoff, clk, true)
	92	+	go copier.copyFrames(daemonToClient)
74	93		}
75	94		startClientCopy()
76	95		startDaemonCopy()

@@ -174,37 +193,37 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
174	193		}
175	194		}
176	195
177		-	func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr, retryEvents chan<- proxyRetryEvent, retryBackoff time.Duration, clk ports.Clock, retryRecoverable bool) {
	196	+	func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
178	197		for {
179		-	f, err := src.Recv()
	198	+	f, err := dir.src.Recv()
180	199		if err != nil {
181		-	errCh <- proxyCopyErr{kind: recvKind, err: err}
	200	+	c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
182	201		return
183	202		}
184	203		retrying := false
185	204		for {
186		-	if err := dst.Send(f); err != nil {
187		-	if retryRecoverable && errors.Is(err, ErrPendingFull) && recoverableClientLink(dst, err) {
	205	+	if err := dir.dst.Send(f); err != nil {
	206	+	if dir.retryPendingFull && errors.Is(err, ErrPendingFull) && recoverableClientLink(dir.dst, err) {
188	207		if !retrying {
189	208		retrying = true
190		-	if !sendProxyRetryEvent(ctx, retryEvents, proxyRetryStart) {
	209	+	if !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryStart) {
191	210		return
192	211		}
193	212		}
194		-	t := clk.NewTimer(retryBackoff)
	213	+	t := c.clk.NewTimer(c.retryBackoff)
195	214		select {
196	215		case <-t.C():
197	216		t.Stop()
198	217		continue
199		-	case <-ctx.Done():
	218	+	case <-c.ctx.Done():
200	219		t.Stop()
201	220		return
202	221		}
203	222		}
204		-	errCh <- proxyCopyErr{kind: sendKind, err: err}
	223	+	c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
205	224		return
206	225		}
207		-	if retrying && !sendProxyRetryEvent(ctx, retryEvents, proxyRetryDone) {
	226	+	if retrying && !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryDone) {
208	227		return
209	228		}
210	229		break

 	proxyRetryDone
 )
+type proxyCopier struct {
+	ctx          context.Context
+	errCh        chan<- proxyCopyErr
+	retryEvents  chan<- proxyRetryEvent
+	retryBackoff time.Duration
+	clk          ports.Clock
+}
+type proxyCopyDirection struct {
+	src              ports.Transport
+	dst              ports.Transport
+	recvKind         proxyCopyErrKind
+	sendKind         proxyCopyErrKind
+	retryPendingFull bool
+}
 func (p ProxyRuntime) Run(ctx context.Context) error {
 	defer func() {
 		_ = p.Client.Close()
 	errCh := make(chan proxyCopyErr, 2)
 	retryEvents := make(chan proxyRetryEvent, 2)
+	copier := proxyCopier{ctx: ctx, errCh: errCh, retryEvents: retryEvents, retryBackoff: retryBackoff, clk: clk}
+	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}
+	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryPendingFull: true}
 	startClientCopy := func() {
-		go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryEvents, retryBackoff, clk, false)
+		go copier.copyFrames(clientToDaemon)
 	}
 	startDaemonCopy := func() {
-		go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryEvents, retryBackoff, clk, true)
+		go copier.copyFrames(daemonToClient)
 	}
 	startClientCopy()
 	startDaemonCopy()
 	}
 }
-func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr, retryEvents chan<- proxyRetryEvent, retryBackoff time.Duration, clk ports.Clock, retryRecoverable bool) {
+func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
 	for {
-		f, err := src.Recv()
+		f, err := dir.src.Recv()
 		if err != nil {
-			errCh <- proxyCopyErr{kind: recvKind, err: err}
+			c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
 			return
 		}
 		retrying := false
 		for {
-			if err := dst.Send(f); err != nil {
+			if err := dir.dst.Send(f); err != nil {
-				if retryRecoverable && errors.Is(err, ErrPendingFull) && recoverableClientLink(dst, err) {
+				if dir.retryPendingFull && errors.Is(err, ErrPendingFull) && recoverableClientLink(dir.dst, err) {
 					if !retrying {
 						retrying = true
-						if !sendProxyRetryEvent(ctx, retryEvents, proxyRetryStart) {
+						if !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryStart) {
 							return
 						}
 					}
-					t := clk.NewTimer(retryBackoff)
+					t := c.clk.NewTimer(c.retryBackoff)
 					select {
 					case <-t.C():
 						t.Stop()
 						continue
-					case <-ctx.Done():
+					case <-c.ctx.Done():
 						t.Stop()
 						return
 					}
 				}
-				errCh <- proxyCopyErr{kind: sendKind, err: err}
+				c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
 				return
 			}
-			if retrying && !sendProxyRetryEvent(ctx, retryEvents, proxyRetryDone) {
+			if retrying && !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryDone) {
 				return
 			}
 			break

M internal/adapters/dgram/transport.go +33 -1


@@ -191,7 +191,39 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
191	191		return nil, err
192	192		}
193	193		opts = normalizeOptions(opts)
194		-	t := &Transport{pc: pc, codec: c, sendDir: sendDir, recvDir: recvDir, mtu: opts.MTU, peer: peer, pending: make(map[uint64]*pending), replay: pdgram.NewReplayWindow(), reasm: pdgram.NewReassembler(), in: make(chan ports.Frame, 32), done: make(chan struct{}), lastHeard: opts.Clock.Now(), heartbeat: opts.Heartbeat, resendAfter: opts.ResendAfter, maxResendAfter: opts.MaxResendAfter, maxResendPerTick: opts.MaxResendPerTick, writeTimeout: opts.WriteTimeout, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, maxPendingWait: opts.MaxPendingWait, maxRecvBuffer: opts.MaxRecvBuffer, clock: opts.Clock, linkState: ports.LinkStateConnected, linkEvents: make(chan ports.LinkEvent, linkEventBufferSize), probeWait: make(map[uint64]chan struct{}), rebind: opts.RebindPacketConn, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
	194	+	t := &Transport{
	195	+	pc: pc,
	196	+	codec: c,
	197	+	sendDir: sendDir,
	198	+	recvDir: recvDir,
	199	+	mtu: opts.MTU,
	200	+	peer: peer,
	201	+	pending: make(map[uint64]*pending),
	202	+	replay: pdgram.NewReplayWindow(),
	203	+	reasm: pdgram.NewReassembler(),
	204	+	in: make(chan ports.Frame, 32),
	205	+	done: make(chan struct{}),
	206	+	lastHeard: opts.Clock.Now(),
	207	+	heartbeat: opts.Heartbeat,
	208	+	resendAfter: opts.ResendAfter,
	209	+	maxResendAfter: opts.MaxResendAfter,
	210	+	maxResendPerTick: opts.MaxResendPerTick,
	211	+	writeTimeout: opts.WriteTimeout,
	212	+	degradedAfter: opts.DegradedAfter,
	213	+	probeAfter: opts.ProbeAfter,
	214	+	offlineAfter: opts.OfflineAfter,
	215	+	deadAfter: opts.DeadAfter,
	216	+	maxPending: opts.MaxPending,
	217	+	maxPendingWait: opts.MaxPendingWait,
	218	+	maxRecvBuffer: opts.MaxRecvBuffer,
	219	+	clock: opts.Clock,
	220	+	linkState: ports.LinkStateConnected,
	221	+	linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
	222	+	probeWait: make(map[uint64]chan struct{}),
	223	+	rebind: opts.RebindPacketConn,
	224	+	nextRecvSeq: 1,
	225	+	recvBuf: make(map[uint64]ports.Frame),
	226	+	}
195	227		t.sendWake = make(chan struct{})
196	228		t.probeReply = make(chan uint64, probeReplyBufferSize)
197	229		t.deliverCond = sync.NewCond(&t.deliverMu)


@@ -191,7 +191,39 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
191		return nil, err	191		return nil, err
192		}	192		}
193		opts = normalizeOptions(opts)	193		opts = normalizeOptions(opts)
194	-	t := &Transport{pc: pc, codec: c, sendDir: sendDir, recvDir: recvDir, mtu: opts.MTU, peer: peer, pending: make(map[uint64]*pending), replay: pdgram.NewReplayWindow(), reasm: pdgram.NewReassembler(), in: make(chan ports.Frame, 32), done: make(chan struct{}), lastHeard: opts.Clock.Now(), heartbeat: opts.Heartbeat, resendAfter: opts.ResendAfter, maxResendAfter: opts.MaxResendAfter, maxResendPerTick: opts.MaxResendPerTick, writeTimeout: opts.WriteTimeout, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, maxPendingWait: opts.MaxPendingWait, maxRecvBuffer: opts.MaxRecvBuffer, clock: opts.Clock, linkState: ports.LinkStateConnected, linkEvents: make(chan ports.LinkEvent, linkEventBufferSize), probeWait: make(map[uint64]chan struct{}), rebind: opts.RebindPacketConn, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}	194	+	t := &Transport{
			195	+	pc: pc,
			196	+	codec: c,
			197	+	sendDir: sendDir,
			198	+	recvDir: recvDir,
			199	+	mtu: opts.MTU,
			200	+	peer: peer,
			201	+	pending: make(map[uint64]*pending),
			202	+	replay: pdgram.NewReplayWindow(),
			203	+	reasm: pdgram.NewReassembler(),
			204	+	in: make(chan ports.Frame, 32),
			205	+	done: make(chan struct{}),
			206	+	lastHeard: opts.Clock.Now(),
			207	+	heartbeat: opts.Heartbeat,
			208	+	resendAfter: opts.ResendAfter,
			209	+	maxResendAfter: opts.MaxResendAfter,
			210	+	maxResendPerTick: opts.MaxResendPerTick,
			211	+	writeTimeout: opts.WriteTimeout,
			212	+	degradedAfter: opts.DegradedAfter,
			213	+	probeAfter: opts.ProbeAfter,
			214	+	offlineAfter: opts.OfflineAfter,
			215	+	deadAfter: opts.DeadAfter,
			216	+	maxPending: opts.MaxPending,
			217	+	maxPendingWait: opts.MaxPendingWait,
			218	+	maxRecvBuffer: opts.MaxRecvBuffer,
			219	+	clock: opts.Clock,
			220	+	linkState: ports.LinkStateConnected,
			221	+	linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
			222	+	probeWait: make(map[uint64]chan struct{}),
			223	+	rebind: opts.RebindPacketConn,
			224	+	nextRecvSeq: 1,
			225	+	recvBuf: make(map[uint64]ports.Frame),
			226	+	}
195		t.sendWake = make(chan struct{})	227		t.sendWake = make(chan struct{})
196		t.probeReply = make(chan uint64, probeReplyBufferSize)	228		t.probeReply = make(chan uint64, probeReplyBufferSize)
197		t.deliverCond = sync.NewCond(&t.deliverMu)	229		t.deliverCond = sync.NewCond(&t.deliverMu)