fix: release a stored session when its PDU session ID is established again

context/datapath.go

@@ -256,6 +256,13 @@ func (node *DataPathNode) DeactivateUpLinkTunnel(smContext *SMContext) {
 			// Remove PDR from PFCP Session
 			smContext.RemovePDRfromPFCPSession(node.UPF.NodeID, pdr)
 
+			if smContext.restoredFromStore {
+				// Its rules were allocated by an earlier incarnation, and this one may have given
+				// the same identifiers to a live session since: freeing them, or removing what the
+				// UPF's pools hold under them, would take them from that session.
+				continue
+			}
+
 			// Remove of UPF
 			err := node.UPF.RemovePDR(pdr)
 			if err != nil {
@@ -303,6 +310,13 @@ func (node *DataPathNode) DeactivateDownLinkTunnel(smContext *SMContext) {
 			// Remove PDR from PFCP Session
 			smContext.RemovePDRfromPFCPSession(node.UPF.NodeID, pdr)
 
+			if smContext.restoredFromStore {
+				// Its rules were allocated by an earlier incarnation, and this one may have given
+				// the same identifiers to a live session since: freeing them, or removing what the
+				// UPF's pools hold under them, would take them from that session.
+				continue
+			}
+
 			// Remove from UPF
 			err := node.UPF.RemovePDR(pdr)
 			if err != nil {
@@ -910,6 +924,10 @@ func (dataPath *DataPath) DeactivateTunnelAndPDR(smContext *SMContext) {
 
 	// Deactivate Tunnels
 	for curDataPathNode := firstDPNode; curDataPathNode != nil; curDataPathNode = curDataPathNode.Next() {
+		if curDataPathNode.UPF == nil {
+			// A data path read back from the datastore can name a UPF this process does not hold.
+			continue
+		}
 		curDataPathNode.DeactivateUpLinkTunnel(smContext)
 		curDataPathNode.DeactivateDownLinkTunnel(smContext)
 	}

context/db.go

@@ -8,6 +8,7 @@
 package context
 
 import (
+	gocontext "context"
 	"encoding/json"
 	"fmt"
 	"hash/fnv"
@@ -64,6 +65,16 @@ func SetupSmfCollection() {
 		logger.DataRepoLog.Errorln("create index failed on TxnId field")
 	}
 
+	// The sessions stored for one subscriber and PDU session ID. Not unique: there are several until
+	// the older ones are released, and one whose delete failed stays stored beside its successor.
+	err = mongoapi.CommonDBClient.EnsureIndex(gocontext.Background(), SmContextDataColl, mongoapi.IndexSpec{
+		Name: smContextKeyIndexName,
+		Keys: mongoapi.AscendingKeys(identifierFilterKey, pduSessionIDFilterKey),
+	})
+	if err != nil {
+		logger.DataRepoLog.Errorf("create index failed on identifier and pduSessionID fields: %v", err)
+	}
+
 	smfCount := mongoapi.CommonDBClient.GetUniqueIdentity("smfCount")
 	logger.DataRepoLog.Infof("unique id - init smfCount %d", smfCount)
 
@@ -127,6 +138,7 @@ func (smContext *SMContext) MarshalJSON() ([]byte, error) {
 		pfcpSessionContextInDB.PDRs = pfcpCtx.PDRs
 		pfcpSessionContextInDB.LocalSEID = SeidConv(pfcpCtx.LocalSEID)
 		pfcpSessionContextInDB.RemoteSEID = SeidConv(pfcpCtx.RemoteSEID)
+		pfcpSessionContextInDB.AcknowledgedAtRecovery = pfcpCtx.AcknowledgedAtRecovery
 		PFCPContextVal[key] = pfcpSessionContextInDB
 	}
 
@@ -187,6 +199,7 @@ func (smContext *SMContext) UnmarshalJSON(data []byte) error {
 			logger.DataRepoLog.Errorf("remoteSEID unmarshal error: %v", err)
 		}
 		smContext.PFCPContext[key].RemoteSEID = remoteSeid
+		smContext.PFCPContext[key].AcknowledgedAtRecovery = pfcpCtxInDB.AcknowledgedAtRecovery
 		// Rebuild the local-SEID index alongside PFCPContext so the SEID-keyed lookups work on a
 		// restored context without iterating the map (see SMContext.seidToPFCPCtx).
 		smContext.recordPFCPCtxRef(localSeid, key, pfcpCtxInDB.NodeID)
@@ -282,6 +295,7 @@ func ToBsonM(data *SMContext) (ret bson.M) {
 		pfcpEntry.PDRs = pfcpCtx.PDRs
 		pfcpEntry.LocalSEID = SeidConv(pfcpCtx.LocalSEID)
 		pfcpEntry.RemoteSEID = SeidConv(pfcpCtx.RemoteSEID)
+		pfcpEntry.AcknowledgedAtRecovery = pfcpCtx.AcknowledgedAtRecovery
 		pfcpContextVal[key] = pfcpEntry
 	}
 
@@ -469,6 +483,9 @@ func GetSMContextByRefInDB(ref string) (smContext *SMContext) {
 		}
 		smContext.RebuildCommunicationClient()
 		smContext.RebuildSMPolicyClient()
+		// Set here, where every context read back from the datastore is built, so that none is
+		// served or released as one this process created.
+		smContext.restoredFromStore = true
 	} else {
 		logger.DataRepoLog.Warnf("SmContext doesn't exist with ref: %v", ref)
 		return nil
@@ -500,7 +517,11 @@ func GetSMContextBySEIDInDB(seidUint uint64) (smContext *SMContext) {
 // DeleteSmContextInDBBySEID Delete SMContext By SEID from DB. Callers (RemoveSMContext) must
 // hold the SMContext's SMLock across this call for the same reason as DeleteSmContextInDBByRef,
 // which this calls.
-func DeleteSmContextInDBBySEID(seidUint uint64) {
+//
+// ref is the session being released. The SEID's document is deleted, and the context it names with
+// it, only while it names that session: the document is keyed by the SEID alone, so it may name
+// another session, and deleting by SEID alone removed that session's records.
+func DeleteSmContextInDBBySEID(seidUint uint64, ref string) {
 	seid := SeidConv(seidUint)
 	logger.DataRepoLog.Infoln("db - delete SMContext In DB by seid")
 	filter := bson.M{seidFilterKey: seid}
@@ -511,7 +532,13 @@ func DeleteSmContextInDBBySEID(seidUint uint64) {
 		logger.DataRepoLog.Warnln(getOneErr)
 	}
 	if result != nil {
-		ref := result[refFilterKey].(string)
+		if mapped, _ := result[refFilterKey].(string); mapped != ref {
+			logger.DataRepoLog.Infof("SEID %v now belongs to SMContext %s, not %s; leaving it", seid, mapped, ref)
+			return
+		}
+		// And the delete is scoped to ref as well, for a session that takes the SEID between the
+		// read above and the delete.
+		filter[refFilterKey] = ref
 
 		delOneErr := mongoapi.CommonDBClient.RestfulAPIDeleteOne(SeidSmContextCol, filter)
 		if delOneErr != nil {

context/db_uptunnel.go

@@ -6,6 +6,7 @@ package context
 
 import (
 	"net"
+	"time"
 
 	gojson "github.com/goccy/go-json"
 	"github.com/omec-project/smf/logger"
@@ -62,6 +63,10 @@ type PFCPSessionContextInDB struct {
 	LocalSEID  string
 	RemoteSEID string
 	NodeID     NodeID
+	// AcknowledgedAtRecovery is kept so that a later start of the SMF can tell whether the node
+	// that acknowledged the session is still the incarnation running. Zero in a document stored
+	// before it was kept.
+	AcknowledgedAtRecovery time.Time
 }
 
 type PFCPContextInDB map[string]PFCPSessionContextInDB

context/export_test.go

@@ -0,0 +1,58 @@
+// SPDX-FileCopyrightText: 2026 Forsway Scandinavia AB
+// SPDX-License-Identifier: Apache-2.0
+
+package context
+
+import "net"
+
+// Exported to the external tests in this directory only.
+
+var StartSmContextWriteWorkers = startSmContextWriteWorkers
+
+// ForgetInMemory drops a context from every in-memory index, leaving only what is stored: the state
+// a restart leaves it in.
+func ForgetInMemory(smContext *SMContext) {
+	smContextPool.Delete(smContext.Ref)
+	canonicalRef.CompareAndDelete(canonicalName(smContext.Identifier, smContext.PDUSessionID), smContext.Ref)
+	for _, pfcpContext := range smContext.PFCPContext {
+		seidSMContextMap.Delete(pfcpContext.LocalSEID)
+	}
+}
+
+// RegisterCurrent makes ref the subscriber's current session for the PDU session ID.
+func RegisterCurrent(identifier string, pduSessionID int32, ref string) {
+	canonicalRef.Store(canonicalName(identifier, pduSessionID), ref)
+}
+
+// SEIDHeldBy is the context the in-memory SEID index holds for a local SEID, without the datastore
+// fallback GetSMContextBySEID adds.
+func SEIDHeldBy(seid uint64) *SMContext {
+	if value, ok := seidSMContextMap.Load(seid); ok {
+		return value.(*SMContext)
+	}
+	return nil
+}
+
+// HoldSEID enters a context in the in-memory SEID index, as establishing its PFCP session does.
+func HoldSEID(seid uint64, smContext *SMContext) {
+	seidSMContextMap.Store(seid, smContext)
+}
+
+// AllocatorHolds reports whether the allocator has ip marked in use.
+func AllocatorHolds(a *IPAllocator, ip net.IP) bool {
+	a.g.lock.Lock()
+	defer a.g.lock.Unlock()
+	return a.g.isUsed[int64(IPAddrOffset(ip.To4(), a.ipNetwork.IP))]
+}
+
+// HoldsQER reports whether the QER with this identifier is still allocated from the UPF's pool.
+func (upf *UPF) HoldsQER(qerID uint32) bool {
+	_, ok := upf.qerPool.Load(qerID)
+	return ok
+}
+
+// HoldsBAR reports whether the BAR with this identifier is still allocated from the UPF's pool.
+func (upf *UPF) HoldsBAR(barID uint8) bool {
+	_, ok := upf.barPool.Load(barID)
+	return ok
+}

context/pfcp_session_context.go

@@ -39,7 +39,8 @@ type PFCPSessionContext struct {
 	// node acknowledged the session, which identifies the incarnation that holds it. A node
 	// refuses an establishment until it is associated, and association is where the held
 	// timestamp is replaced, so a session a restarted node accepted carries the new one. Zero when
-	// unknown -- a session restored from the database -- which restoration treats as lost.
+	// unknown -- a session restored from a document stored before it was kept -- which restoration
+	// treats as lost.
 	AcknowledgedAtRecovery time.Time
 }
 

context/sm_context.go

@@ -358,6 +358,11 @@ type SMContext struct {
 	// disconnect event, so without this the final Kafka event would report an empty UPF.
 	lastUpfName string `json:"-" yaml:"-" bson:"-"`
 	lastUpfIP   string `json:"-" yaml:"-" bson:"-"`
+
+	// restoredFromStore is set on a context read back from the datastore rather than created by this
+	// process. Such a session was never counted in the session gauge, and the rule identifiers it
+	// carries were allocated by an earlier incarnation, so releasing it frees none of them here.
+	restoredFromStore bool `json:"-" yaml:"-" bson:"-"`
 }
 
 func canonicalName(identifier string, pduSessID int32) (canonical string) {
@@ -676,8 +681,10 @@ func GetSMContext(ref string) (smContext *SMContext) {
 		// could not delete: without that check, a pool miss right after a failed delete would
 		// read the still-present, released document back from Mongo and resurrect it here.
 		if dbContext := GetSMContextByRefInDB(ref); dbContext != nil {
-			smContextPool.Store(ref, dbContext)
-			smContext = dbContext
+			// LoadOrStore: another lookup, or a release of the stored session, may have put an
+			// instance in the pool since the miss, and both have to act on that one.
+			actual, _ := smContextPool.LoadOrStore(ref, dbContext)
+			smContext = actual.(*SMContext)
 		}
 	}
 
@@ -891,9 +898,11 @@ func RemoveSMContextLocked(smContext *SMContext) {
 	smContext.DeferredPolicyDecisions = nil
 
 	for _, pfcpSessionContext := range smContext.PFCPContext {
-		seidSMContextMap.Delete(pfcpSessionContext.LocalSEID)
+		// Only this session's own entries: the index is keyed by the SEID alone, so an entry may be
+		// another session's.
+		seidSMContextMap.CompareAndDelete(pfcpSessionContext.LocalSEID, smContext)
 		if factory.SmfConfig.Configuration.EnableDbStore {
-			DeleteSmContextInDBBySEID(pfcpSessionContext.LocalSEID)
+			DeleteSmContextInDBBySEID(pfcpSessionContext.LocalSEID, smContext.Ref)
 		}
 	}
 
@@ -920,9 +929,13 @@ func RemoveSMContextLocked(smContext *SMContext) {
 	// Use CompareAndDelete so a replacement context (created after this one was superseded via
 	// restoration) is not accidentally unlinked from the canonical map.
 	canonicalRef.CompareAndDelete(canonicalName(smContext.Identifier, smContext.PDUSessionID), smContext.Ref)
-	// Sess Stats
-	smContextActive := decSMContextActive()
-	metrics.SetSessStats(SMF_Self().NfInstanceID, smContextActive)
+	// Sess Stats. A context read back from the datastore was never counted -- only NewSMContext
+	// counts -- and decrementing for it would undercount the live sessions by one, or wrap the
+	// unsigned count when none is counted.
+	if !smContext.restoredFromStore {
+		smContextActive := decSMContextActive()
+		metrics.SetSessStats(SMF_Self().NfInstanceID, smContextActive)
+	}
 }
 
 // *** add unit test ***//

context/stored_session.go

@@ -0,0 +1,102 @@
+// SPDX-FileCopyrightText: 2026 Forsway Scandinavia AB
+// SPDX-License-Identifier: Apache-2.0
+
+package context
+
+import (
+	gojson "github.com/goccy/go-json"
+	"github.com/omec-project/smf/logger"
+	"github.com/omec-project/util/mongoapi"
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+// A session read back from the datastore after this process started is not one this process
+// created. When its subscriber establishes the same PDU session again it must be released, as one
+// held in memory is (TS 24.501 clause 6.4.1.7 c)).
+
+const (
+	identifierFilterKey   = "identifier"
+	pduSessionIDFilterKey = "pduSessionID"
+	smContextKeyIndexName = "identifier_pduSessionID"
+)
+
+// ReleaseRestoredSession releases a session built from the datastore: its stored documents and its
+// PFCP sessions at the user planes this process holds. The producer installs the full release,
+// which can send PFCP; the default removes the context only.
+var ReleaseRestoredSession = func(smContext *SMContext) {
+	smContext.SMLock.Lock()
+	defer smContext.SMLock.Unlock()
+	RemoveSMContextLocked(smContext)
+}
+
+// storedRefs lists the references of the sessions stored for one subscriber and PDU session ID. Each
+// document is decoded through JSON, as GetSMContextByRefInDB decodes one.
+func storedRefs(identifier string, pduSessionID int32) ([]string, error) {
+	// The filter's PDU session ID is an int32 and the write path stores the field as a JSON number;
+	// the datastore compares numbers by value whatever their stored type, so the two match.
+	filter := bson.M{identifierFilterKey: identifier, pduSessionIDFilterKey: pduSessionID}
+	docs, err := mongoapi.CommonDBClient.RestfulAPIGetMany(SmContextDataColl, filter)
+	if err != nil {
+		return nil, err
+	}
+	refs := make([]string, 0, len(docs))
+	for _, doc := range docs {
+		var s struct {
+			Ref string `json:"ref"`
+		}
+		if err := gojson.Unmarshal(mapToByte(doc), &s); err != nil || s.Ref == "" {
+			logger.DataRepoLog.Warnf("stored session for %s-%d could not be read: %v", identifier, pduSessionID, err)
+			continue
+		}
+		refs = append(refs, s.Ref)
+	}
+	return refs, nil
+}
+
+// restoredInstance returns the context for a stored ref, for releasing it: the instance in the pool,
+// so the release and anything else using it act on the same session. On a pool miss it is read from
+// the datastore and put in the pool, unless a lookup has put one there meanwhile, which is then the
+// one returned: the release deletes the pool's entry for ref, which must not be another instance than
+// the one it released. Nil when the document is gone or its delete failed earlier.
+func restoredInstance(ref string) *SMContext {
+	if value, ok := smContextPool.Load(ref); ok {
+		return value.(*SMContext)
+	}
+	if IsSmContextDeleteFailed(ref) {
+		return nil
+	}
+	smContext := GetSMContextByRefInDB(ref)
+	if smContext == nil {
+		return nil
+	}
+	actual, _ := smContextPool.LoadOrStore(ref, smContext)
+	return actual.(*SMContext)
+}
+
+// ReleaseStoredSessions releases every session stored for a subscriber and PDU session ID, before a
+// new session is established with that ID. It is called when the ID resolves to no session in
+// memory: after a restart the previous one is in the datastore only, and TS 24.501 clause 6.4.1.7 c)
+// requires it to be released as surely as one held in memory. A stored session that a lookup has
+// already read into the pool is released as that instance.
+func ReleaseStoredSessions(identifier string, pduSessionID int32) {
+	refs, err := storedRefs(identifier, pduSessionID)
+	if err != nil {
+		logger.CtxLog.Warnf("stored sessions for %s-%d could not be listed: %v", identifier, pduSessionID, err)
+		return
+	}
+	for _, ref := range refs {
+		smContext := restoredInstance(ref)
+		if smContext == nil {
+			continue
+		}
+		smContext.SubPduSessLog.Warn("PDUSessionSMContextCreate, stored session with this PDU session ID exists, releasing it")
+		ReleaseRestoredSession(smContext)
+	}
+}
+
+// RestoredFromStore reports whether the context was read back from the datastore after this process
+// started. It is set before the context is published to the pool and never changes, so it can be
+// read without SMLock.
+func (smContext *SMContext) RestoredFromStore() bool {
+	return smContext.restoredFromStore
+}