refactor(peer): run Manager connections through the protocol core

diam/sm/internal/parserbridge/parserbridge.go → diam/internal/parserbridge/parserbridge.go

@@ -1,4 +1,4 @@
-// Package parserbridge gives package sm the smparser conversions that are not
+// Package parserbridge gives the protocol adapters the smparser conversions that are not
 // part of smparser's public API. smparser installs each one during its
 // initialization; this package cannot import smparser, so each is held as any
 // with its concrete function type documented beside it.

diam/internal/protocol/accessors.go

@@ -1,8 +1,33 @@
 package protocol
 
+import (
+	"github.com/gomaja/go-diameter/diam"
+	"github.com/gomaja/go-diameter/diam/internal/watchdog"
+)
+
 // Phase returns the current phase within the caller's serialization domain.
 func (c *Core) Phase() Phase { return c.phase }
 
 // Ready reports whether admission callbacks have completed within the caller's
 // serialization domain.
 func (c *Core) Ready() bool { return c.ready }
+
+// Progress carries scalar protocol state for executors that publish peer
+// eligibility. Capability/error evidence is copied only by Snapshot.
+type Progress struct {
+	Phase    Phase
+	Ready    bool
+	Watchdog watchdog.Machine
+	Timers   [timerCount]TimerSnapshot
+	Controls [3]ControlSnapshot
+}
+
+func (c *Core) Progress() Progress {
+	p := Progress{Phase: c.phase, Ready: c.ready, Watchdog: c.wd, Timers: c.timers}
+	for i, command := range [...]uint32{diam.CapabilitiesExchange, diam.DeviceWatchdog, diam.DisconnectPeer} {
+		if r := c.controls[command]; r != nil {
+			p.Controls[i] = ControlSnapshot{Command: r.command, Reservation: r.reservation, Logical: r.logical, Answered: r.answered, Writers: r.writers}
+		}
+	}
+	return p
+}

diam/internal/protocol/candidate_election_test.go

@@ -0,0 +1,43 @@
+package protocol
+
+import (
+	"testing"
+	"time"
+
+	"github.com/gomaja/go-diameter/diam"
+	"github.com/gomaja/go-diameter/diam/internal/watchdog"
+)
+
+func TestCandidateActivationUsesPeerRecoveryHistory(t *testing.T) {
+	c := machine(t, func(cfg *Config) { cfg.Watchdog = true; cfg.WatchdogInterval = 6 * time.Second })
+	c.Handle(event(Start))
+	candidate := find(t, c.Handle(receive(cer(t))), ValidatedCER)
+	in := event(Activate)
+	in.Token, in.Recovery = candidate.Token, true
+	sent := find(t, c.Handle(in), Send)
+	ready := find(t, c.Handle(complete(sent)), HandshakeReady)
+	in = event(ReleaseAdmission)
+	in.Token = ready.Token
+	probe := grant(t, c, c.Handle(in))
+	if probe.Message.Header.CommandCode != diam.DeviceWatchdog || c.Snapshot().Watchdog().State != watchdog.Reopen {
+		t.Fatal("recovered accepted peer skipped REOPEN")
+	}
+}
+
+func TestCandidateTransfersDeadlineToElectionOwner(t *testing.T) {
+	c := machine(t, func(cfg *Config) { cfg.CandidateDeadlineOwnedByPeer = true })
+	c.Handle(event(Start))
+	before := c.Snapshot().Timer(HandshakeTimer)
+	es := c.Handle(receive(cer(t)))
+	find(t, es, ValidatedCER)
+	if c.Snapshot().Timer(HandshakeTimer).Token != 0 {
+		t.Fatal("candidate retained the pre-CER deadline")
+	}
+	in := event(TimerExpired)
+	in.Timer = HandshakeTimer
+	in.Token = before.Token
+	in.At = before.Deadline
+	if got := c.Handle(in); len(got) != 0 || c.Snapshot().Candidate() == 0 {
+		t.Fatal("pre-CER timeout cancelled the election candidate")
+	}
+}

diam/internal/protocol/closing.go

@@ -7,9 +7,9 @@ import (
 	"github.com/gomaja/go-diameter/diam/internal/base"
 )
 
-// beginClosing latches before DPR/DPA effects. Frozen acknowledges the #125
-// native write-gate boundary, not queue selection (RFC 6733 §§5.4,5.6).
-func (c *Core) beginClosing(in Input, local bool) []Effect {
+// beginClosing latches before DPR/DPA effects. Frozen acknowledges the native
+// write-gate boundary before selecting disconnect control.
+func (c *Core) beginClosing(es []Effect, in Input, local bool) []Effect {
 	c.phase = Closing
 	c.ready = false
 	c.closeLocal = local
@@ -20,45 +20,45 @@ func (c *Core) beginClosing(in Input, local bool) []Effect {
 		c.disconnectWaiting = true
 		c.disconnectToken = in.Token
 	}
-	es := c.stop(WatchdogTimer)
+	es = c.stop(es, WatchdogTimer)
 	duration := in.Timeout
 	if duration <= 0 {
 		duration = c.cfg.ClosingTimeout
 	}
-	es = append(es, c.arm(ClosingTimer, in.At, duration)...)
+	es = c.arm(es, ClosingTimer, in.At, duration)
 	e := c.effect(FreezeAdmission)
 	e.Token = c.fenceToken
 	return append(es, e)
 }
-func (c *Core) freeze(in Input) []Effect {
+func (c *Core) freeze(es []Effect, in Input) []Effect {
 	if c.phase != Closing || c.frozen || in.Token != c.fenceToken {
-		return nil
+		return es
 	}
 	c.frozen = true
 	c.fenceToken = 0
 	if c.closeLocal {
-		return c.reserve(diam.DisconnectPeer)
+		return c.reserve(es, diam.DisconnectPeer)
 	}
 	saved := c.closeRequest
 	c.closeRequest = Input{}
 	saved.InFlight = in.InFlight
-	return c.answerDPR(saved, true)
+	return c.answerDPR(es, saved, true)
 }
-func (c *Core) disconnect(in Input) []Effect {
+func (c *Core) disconnect(es []Effect, in Input) []Effect {
 	if c.disconnectWaiting {
-		return []Effect{c.notify(Notification{Kind: DisconnectNotice, Outcome: AlreadyPending})}
+		return append(es, c.notify(Notification{Kind: DisconnectNotice, Outcome: AlreadyPending}))
 	}
 	// Preserve StateMachine's local API after a received DPR (RFC 6733
 	// §§5.4.1,5.6). Neither the original receiver deadline nor admission changes.
 	if c.cfg.LocalDisconnectInClosing && c.phase == Closing && c.frozen && !c.closeLocal && in.Cause <= 2 {
 		c.disconnectWaiting, c.disconnectToken, c.closeCause = true, in.Token, in.Cause
-		es := c.arm(ReceiverDisconnectTimer, in.At, in.Timeout)
-		return append(es, c.reserve(diam.DisconnectPeer)...)
+		es = c.arm(es, ReceiverDisconnectTimer, in.At, in.Timeout)
+		return c.reserve(es, diam.DisconnectPeer)
 	}
-	if c.phase != Open || in.Cause > 2 {
-		return []Effect{c.notify(Notification{Kind: DisconnectNotice, Outcome: NotOpen})}
+	if c.phase != Open || in.Cause > 2 && !c.cfg.AllowAnyLocalDisconnectCause {
+		return append(es, c.notify(Notification{Kind: DisconnectNotice, Outcome: NotOpen}))
 	}
-	return c.beginClosing(in, true)
+	return c.beginClosing(es, in, true)
 }
 func (c *Core) bound(m *diam.Message, realm bool) bool {
 	var identity base.DWR
@@ -67,18 +67,19 @@ func (c *Core) bound(m *diam.Message, realm bool) bool {
 	}
 	return equalIdentity(string(identity.OriginHost), c.caps.Host) && (!realm || equalIdentity(string(identity.OriginRealm), c.caps.Realm))
 }
-func (c *Core) dpr(in Input) (es []Effect) {
+func (c *Core) dpr(es []Effect, in Input) (out []Effect) {
 	cause, err := base.ValidateDPR(in.Message)
 	if err == nil && !c.bound(in.Message, false) {
 		err = errors.New("DPR Origin-Host does not match the admitted peer (RFC 6733 §6.3)")
 	}
 	if err != nil {
-		return c.discardCause(in, diam.ProtocolInvalidDPR, err)
+		return c.discardCause(es, in, diam.ProtocolInvalidDPR, err)
 	}
 	in.Cause = cause
+	start := len(es)
 	defer func() {
-		if len(es) != 0 {
-			es[0].ReleaseDispatch = true
+		if len(out) > start {
+			out[start].ReleaseDispatch = true
 		}
 	}()
 	// RFC 6733 §5.4.3: reconnect suppression is decided by the validated
@@ -90,49 +91,49 @@ func (c *Core) dpr(in Input) (es []Effect) {
 		es = append(es, e)
 	}
 	if c.phase != Closing {
-		es = append(es, c.validActivity()...)
-		return append(es, c.beginClosing(in, false)...)
+		es = c.validActivity(es)
+		return c.beginClosing(es, in, false)
 	}
 	if !c.frozen {
-		return append(es, c.discard(in, "", 0)...)
+		return c.discard(es, in, "", 0)
 	}
-	return append(es, c.answerDPR(in, !c.peerDisconnectSelected)...)
+	return c.answerDPR(es, in, !c.peerDisconnectSelected)
 }
-func (c *Core) answerDPR(in Input, first bool) []Effect {
+func (c *Core) answerDPR(es []Effect, in Input, first bool) []Effect {
 	result := uint32(diam.Success)
 	if in.InFlight {
 		result = diam.UnableToComply
 	}
 	answer, err := base.BuildDPAResult(in.Message, c.cfg.Local, result)
 	if err != nil {
-		return c.fail(in, diam.ProtocolReplyFailure, result)
+		return c.fail(es, in, diam.ProtocolReplyFailure, result)
 	}
 	var notice *Notification
 	if first {
 		c.peerDisconnectSelected = true
 		notice = &Notification{Kind: PeerDisconnectNotice, Cause: c.peerCause, ReplyCode: result, Message: metadata(in.Message)}
 	}
-	return c.send(answer, nil, true, false, false, notice, in.Message)
+	return c.send(es, answer, nil, true, false, false, notice, in.Message)
 }
-func (c *Core) dpa(in Input) []Effect {
+func (c *Core) dpa(es []Effect, in Input) []Effect {
 	r := c.controls[diam.DisconnectPeer]
 	if !c.matches(r, in.Message) {
-		es := c.discard(in, diam.ProtocolUnexpectedMessage, 0)
+		es = c.discard(es, in, diam.ProtocolUnexpectedMessage, 0)
 		if r != nil && r.logical {
 			es[len(es)-1].Notification.Diagnostic = "dpa-correlation"
 		}
 		return es
 	}
 	result, err := base.ValidateDPA(in.Message)
 	if err != nil {
-		return c.discardCause(in, diam.ProtocolInvalidDPA, err)
+		return c.discardCause(es, in, diam.ProtocolInvalidDPA, err)
 	}
 	if !c.bound(in.Message, false) {
-		es := c.discard(in, diam.ProtocolUnexpectedMessage, 0)
+		es = c.discard(es, in, diam.ProtocolUnexpectedMessage, 0)
 		es[len(es)-1].Notification.Diagnostic = "dpa-host"
 		return es
 	}
-	es := c.settle(r)
+	es = c.settle(es, r)
 	// Synchronous frontends complete their DPR send before publishing the
 	// disconnect result. An early answer settles ownership without cancelling
 	// the original Closing deadline or the selected writer (#110 API boundary).
@@ -143,14 +144,14 @@ func (c *Core) dpa(in Input) []Effect {
 		c.earlyDPA = &saved
 		return es
 	}
-	return append(es, c.completeDPA(in, result)...)
+	return c.completeDPA(es, in, result)
 }
 
-func (c *Core) completeDPA(in Input, result uint32) []Effect {
+func (c *Core) completeDPA(es []Effect, in Input, result uint32) []Effect {
 	c.earlyDPA = nil
-	var es []Effect
+
 	c.disconnectWaiting = false
-	es = append(es, c.terminate(Closed)...)
+	es = c.terminate(es, Closed)
 	n := Notification{Kind: DisconnectNotice, ResultCode: result, Outcome: Completed}
 	e := c.notify(n)
 	e.Token = c.disconnectToken