Fixes after restoration (#637)

* Stale value in SBIPFCPCommunicationChan after restoration

Signed-off-by: Arrobo, Gabriel <[email protected]>

* Added a recover() around the goroutine body

Signed-off-by: Arrobo, Gabriel <[email protected]>

* Remove duplicate logging

Signed-off-by: Arrobo, Gabriel <[email protected]>

* Address Copilot's comments

Signed-off-by: Arrobo, Gabriel <[email protected]>

---------

Signed-off-by: Arrobo, Gabriel <[email protected]>

Author: gab-arrobo

consumer/nf_management.go

@@ -747,12 +747,6 @@ func SendCreateSubscription(nrfUri string, nrfSubscriptionData models.Subscripti
 	if err == nil {
 		return nrfSubData, nil, nil
 	} else if res != nil {
-		// Logged for every error response now, not only for the ones the removed guard let
-		// through. Warn rather than Error: with the guard gone, a response whose body decodes is
-		// returned to the caller as problem details and a nil error, so the caller decides whether
-		// that is an error and reports it. Logging it as one here would contradict the nil error
-		// and duplicate the caller's own line.
-		logger.ConsumerLog.Warnf("SendCreateSubscription received error response: %v", res.Status)
 		if problem, handledErr := util.HandleOpenAPIError(err); problem != nil {
 			return nrfSubData, problem, nil
 		} else if handledErr != nil {

pfcp/adapter/adapter.go

@@ -323,12 +323,20 @@ func HandlePfcpSessionEstablishmentResponse(msg *udp.Message) {
 			logger.PfcpLog.Errorf("pfcp session establishment response cause error: %v", err)
 			return
 		}
+		// Gated on the state, like the modification and release handlers. Restoration issues an
+		// establishment without waiting on this channel, so an unconditional send here would leave a
+		// stale value for whichever unrelated modification or release next waits on it.
+		awaited := smContext.SMContextState == context.SmStatePfcpCreatePending
 		// UPF Accept
 		if causeValue == ie.CauseRequestAccepted {
-			smContext.SBIPFCPCommunicationChan <- context.SessionEstablishSuccess
+			if awaited {
+				smContext.SBIPFCPCommunicationChan <- context.SessionEstablishSuccess
+			}
 			smContext.SubPfcpLog.Infof("PFCP Session Establishment accepted")
 		} else {
-			smContext.SBIPFCPCommunicationChan <- context.SessionEstablishFailed
+			if awaited {
+				smContext.SBIPFCPCommunicationChan <- context.SessionEstablishFailed
+			}
 			smContext.SubPfcpLog.Errorf("PFCP Session Establishment rejected with cause [%v]", causeValue)
 			if causeValue == ie.CauseNoEstablishedPFCPAssociation {
 				SetUpfInactive(*rspNodeID)

pfcp/handler/handler.go

@@ -538,11 +538,19 @@ func HandlePfcpSessionEstablishmentResponse(msg *udp.Message) {
 			logger.PfcpLog.Errorf("failed to parse Cause IE: %+v", err)
 			return
 		}
+		// Gated on the state, like the modification and release handlers below. Restoration issues
+		// an establishment without waiting on this channel, so an unconditional send here would leave
+		// a stale value for whichever unrelated modification or release next waits on it.
+		awaited := smContext.SMContextState == smf_context.SmStatePfcpCreatePending
 		if causeValue == ie.CauseRequestAccepted {
-			smContext.SBIPFCPCommunicationChan <- smf_context.SessionEstablishSuccess
+			if awaited {
+				smContext.SBIPFCPCommunicationChan <- smf_context.SessionEstablishSuccess
+			}
 			smContext.SubPfcpLog.Infoln("PFCP Session Establishment accepted")
 		} else {
-			smContext.SBIPFCPCommunicationChan <- smf_context.SessionEstablishFailed
+			if awaited {
+				smContext.SBIPFCPCommunicationChan <- smf_context.SessionEstablishFailed
+			}
 			smContext.SubPfcpLog.Errorf("PFCP Session Establishment rejected with cause [%v]", causeValue)
 			if causeValue == ie.CauseNoEstablishedPFCPAssociation {
 				SetUpfInactive(*rspNodeID, msg.PfcpMessage.MessageTypeName())

pfcp/handler/handler_test.go

@@ -284,3 +284,113 @@ func TestHandlePfcpSessionEstablishmentResponseNilTunnel(t *testing.T) {
 		t.Errorf("expected pending PFCP txn for seq %d to be consumed on the nil-Tunnel ignore path, but it was still present", seq)
 	}
 }
+
+// TestHandlePfcpSessionEstablishmentResponseChannelGatedByState covers the SMContextState gate on
+// SBIPFCPCommunicationChan: the normal establishment path waits in SmStatePfcpCreatePending and
+// must receive the signal, while restoration's reissue leaves the context in some other state and
+// must not receive it (an unconditional send would leave a stale value for the next unrelated
+// modification or release that waits on the channel).
+func TestHandlePfcpSessionEstablishmentResponseChannelGatedByState(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		imsi       string
+		state      context.SMContextState
+		wantSignal bool
+	}{
+		{
+			name:       "awaited establishment sends the signal",
+			imsi:       "imsi-100000000000001",
+			state:      context.SmStatePfcpCreatePending,
+			wantSignal: true,
+		},
+		{
+			name:       "unawaited response (e.g. restoration) withholds the signal",
+			imsi:       "imsi-100000000000002",
+			state:      context.SmStateActive,
+			wantSignal: false,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			// AllocateLocalSEID reads factory.SmfConfig.Configuration.EnableDbStore, so the config
+			// must be initialized for the SEID allocation path not to panic when this test runs in
+			// isolation.
+			if factory.SmfConfig.Configuration == nil {
+				factory.SmfConfig = factory.Config{
+					Configuration: &factory.Configuration{
+						KafkaInfo:        factory.KafkaInfo{EnableKafka: boolPointer(false)},
+						EnableUpfAdapter: false,
+					},
+				}
+			}
+
+			nodeID := context.NewNodeID("1.1.1.1")
+			smContext := context.NewSMContext(tc.imsi, 10)
+			smContext.SMContextState = tc.state
+
+			smContext.Tunnel = &context.UPTunnel{
+				DataPathPool: context.DataPathPool{
+					10: &context.DataPath{
+						IsDefaultPath: true,
+						FirstDPNode: &context.DataPathNode{
+							UPF: &context.UPF{NodeID: *nodeID},
+						},
+					},
+				},
+			}
+
+			datapath := &context.DataPath{
+				FirstDPNode: &context.DataPathNode{
+					UPF: &context.UPF{NodeID: *nodeID},
+				},
+			}
+			smContext.AllocateLocalSEIDForDataPath(datapath)
+
+			var localSEID uint64
+			for _, pfcpCtx := range smContext.PFCPContext {
+				if pfcpCtx.LocalSEID != 0 {
+					localSEID = pfcpCtx.LocalSEID
+				}
+			}
+			if localSEID == 0 {
+				t.Fatal("failed to allocate a local SEID for the test SMContext")
+			}
+
+			seq := uint32(localSEID)
+			pfcp_message.InsertPfcpTxn(seq, nodeID)
+
+			rsp := message.NewSessionEstablishmentResponse(
+				0,
+				0,
+				localSEID,
+				seq,
+				0,
+				ie.NewCause(ie.CauseRequestAccepted),
+				ie.NewNodeID("1.1.1.1", "", ""),
+				ie.NewRecoveryTimeStamp(time.Now()),
+			)
+
+			udpMessage := udp.Message{
+				RemoteAddr: &net.UDPAddr{
+					IP:   net.ParseIP("1.1.1.1"),
+					Port: 8809,
+				},
+				PfcpMessage: rsp,
+			}
+
+			handler.HandlePfcpSessionEstablishmentResponse(&udpMessage)
+
+			select {
+			case status := <-smContext.SBIPFCPCommunicationChan:
+				if !tc.wantSignal {
+					t.Errorf("expected no send to SBIPFCPCommunicationChan when SMContextState is %v, got signal %v", tc.state, status)
+				} else if status != context.SessionEstablishSuccess {
+					t.Errorf("expected SessionEstablishSuccess, got %v", status)
+				}
+			default:
+				if tc.wantSignal {
+					t.Error("expected a send to SBIPFCPCommunicationChan when SMContextState is SmStatePfcpCreatePending, got none")
+				}
+			}
+		})
+	}
+}

producer/restoration.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -182,6 +183,14 @@ func RestoreSessionsOnUPF(nodeID context.NodeID, recovery time.Time) {
 
 	go func() {
 		defer restorationsInProgress.CompareAndDelete(nodeIP, run)
+		// A panic here must not escape: this repairs one restarted UPF on a background goroutine,
+		// and an unrecovered panic in any goroutine takes down the whole process -- turning one UPF's
+		// restart into an SMF restart that loses every session on every UPF, not just this one.
+		defer func() {
+			if r := recover(); r != nil {
+				logger.PfcpLog.Errorf("UPF[%s] restoration panicked and was abandoned: %v\n%s", nodeIP, r, debug.Stack())
+			}
+		}()
 		restoreSessions(nodeID, nodeIP, run)
 	}()
 }
@@ -804,7 +813,7 @@ func resolveUnrestorable(unrestored []*context.SMContext, nodeIP string) (notRel
 func releaseOneSession(smContext *context.SMContext) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
-			err = fmt.Errorf("releasing the session panicked: %v", r)
+			err = fmt.Errorf("releasing the session panicked: %v\n%s", r, debug.Stack())
 		}
 	}()
 	return releaseUnrestorableSession(smContext)