Merge pull request #540 from openziti/fix.api.create.during.sync

do not remove api sessions added after/during sync

controller/handler_edge_ctrl/heartbeat.go

@@ -38,17 +38,34 @@ func (h *sessionHeartbeatHandler) ContentType() int32 {
 	return env.ApiSessionHeartbeatType
 }
 
-func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
+func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
 	go func() {
 		req := &edge_ctrl_pb.ApiSessionHeartbeat{}
+		routerId := ch.Id().Token
 		if err := proto.Unmarshal(msg.Body, req); err == nil {
 
-			err := h.appEnv.GetHandlers().ApiSession.MarkActivity(req.Tokens)
+			notFoundTokens, err := h.appEnv.GetHandlers().ApiSession.MarkActivity(req.Tokens)
 
 			if err != nil {
 				pfxlog.Logger().
 					WithError(err).
-					Error("unable to set activity for heartbeat")
+					WithField("routerId", routerId).
+					Errorf("unable to set activity for heartbeat: %v", err)
+			}
+
+			if len(notFoundTokens) > 0 {
+				pfxlog.Logger().
+					WithField("routerId", routerId).
+					Debugf("api session tokens not found during heartbeat, sending delete: %v", notFoundTokens)
+
+				msgStruct := &edge_ctrl_pb.ApiSessionRemoved{
+					Tokens: notFoundTokens,
+				}
+
+				msgBytes, _ := proto.Marshal(msgStruct)
+				msg := channel2.NewMessage(env.ApiSessionRemovedType, msgBytes)
+
+				_ = ch.Send(msg)
 			}
 
 		} else {

controller/model/api_session_handlers.go

@@ -90,23 +90,40 @@ func (handler *ApiSessionHandler) Delete(id string) error {
 	return handler.deleteEntity(id)
 }
 
-func (handler *ApiSessionHandler) MarkActivity(tokens []string) error {
-	return handler.GetDb().Batch(func(tx *bbolt.Tx) error {
+// MarkActivity returns tokens that were not found if any and/or an error.
+func (handler *ApiSessionHandler) MarkActivity(tokens []string) ([]string, error) {
+	var notFoundTokens []string
+
+	err := handler.GetDb().Batch(func(tx *bbolt.Tx) error {
 		store := handler.Store.(persistence.ApiSessionStore)
 		mutCtx := boltz.NewMutateContext(tx)
 		for _, token := range tokens {
 			apiSession, err := store.LoadOneByToken(tx, token)
 			if err != nil {
-				return err
+				if boltz.IsErrNotFoundErr(err) {
+					notFoundTokens = append(notFoundTokens, token)
+				} else {
+					return err
+				}
+
 			}
 			if err = store.Update(mutCtx, apiSession, persistence.UpdateTimeOnlyFieldChecker{}); err != nil {
-				return err
+				if err != nil {
+					if boltz.IsErrNotFoundErr(err) {
+						notFoundTokens = append(notFoundTokens, token)
+					} else {
+						return err
+					}
+
+				}
 			}
 
 			handler.env.GetHandlers().Identity.SetActive(apiSession.IdentityId)
 		}
 		return nil
 	})
+
+	return notFoundTokens, err
 }
 
 func (handler *ApiSessionHandler) Stream(query string, collect func(*ApiSession, error) error) error {

router/handler_edge_ctrl/apiSessionAdded.go

@@ -37,7 +37,7 @@ type apiSessionAddedHandler struct {
 	syncTracker *apiSessionSyncTracker
 
 	reqChan   chan *apiSessionAddedWithState
-	syncReady chan []*edge_ctrl_pb.ApiSession
+	syncReady chan *apiSessionSyncTracker
 	syncFail  chan error
 
 	stop chan struct{}
@@ -49,7 +49,7 @@ func NewApiSessionAddedHandler(sm fabric.StateManager, control channel2.Channel)
 		sm:      sm,
 		reqChan: make(chan *apiSessionAddedWithState, 100),
 
-		syncReady: make(chan []*edge_ctrl_pb.ApiSession, 0),
+		syncReady: make(chan *apiSessionSyncTracker, 0),
 		syncFail:  make(chan error, 0),
 
 		stop: make(chan struct{}, 0),
@@ -109,13 +109,20 @@ func (h *apiSessionAddedHandler) startSyncApplier() {
 		select {
 		case <-h.stop:
 			return
-		case apiSessions := <-h.syncReady:
+		case sync := <-h.syncReady:
+			lastId := ""
+			apiSessions := sync.all()
 			for _, apiSession := range apiSessions {
+				if lastId == "" || apiSession.Id > lastId {
+					lastId = apiSession.Id
+				}
+
 				h.sm.AddApiSession(apiSession)
 			}
-			h.sm.RemoveMissingApiSessions(apiSessions)
 
-			pfxlog.Logger().Infof("finished sychronizing api sessions [count: %d]", len(apiSessions))
+			h.sm.RemoveMissingApiSessions(apiSessions, lastId)
+
+			pfxlog.Logger().Infof("finished sychronizing api sessions [count: %d, syncId: %s]", len(apiSessions), sync.syncId)
 		}
 	}
 }
@@ -149,7 +156,7 @@ func (h *apiSessionAddedHandler) legacySync(reqWithState *apiSessionAddedWithSta
 		h.sm.AddApiSession(apiSession)
 	}
 
-	h.sm.RemoveMissingApiSessions(reqWithState.ApiSessions)
+	h.sm.RemoveMissingApiSessions(reqWithState.ApiSessions, "")
 }
 
 func (h *apiSessionAddedHandler) startReceiveSync() {
@@ -188,7 +195,7 @@ func (h *apiSessionAddedHandler) instantSync(reqWithState *apiSessionAddedWithSt
 
 		if h.syncTracker == nil || h.syncTracker.syncId == "" {
 			logger.Infof("first api session syncId [%s], starting", reqWithState.Id)
-		} else if h.syncTracker.isDone{
+		} else if h.syncTracker.isDone {
 			logger.Infof("api session syncId [%s], starting", reqWithState.Id)
 		} else {
 			logger.Infof("api session with newer syncId [old: %s, new: %s], aborting old, starting new", h.syncTracker.syncId, reqWithState.Id)
@@ -214,13 +221,13 @@ func (h *apiSessionAddedHandler) instantSync(reqWithState *apiSessionAddedWithSt
 }
 
 type apiSessionSyncTracker struct {
-	syncId           string
-	reqsWithState    map[int]*apiSessionAddedWithState
-	hasLast          bool
-	lastSeq          int
-	stop             chan struct{}
-	deadline         sync.Once
-	isDone           bool
+	syncId        string
+	reqsWithState map[int]*apiSessionAddedWithState
+	hasLast       bool
+	lastSeq       int
+	stop          chan struct{}
+	deadline      sync.Once
+	isDone        bool
 }
 
 func newApiSessionSyncTracker(id string) *apiSessionSyncTracker {
@@ -251,7 +258,7 @@ func (tracker *apiSessionSyncTracker) IsDone() bool {
 	return tracker.isDone
 }
 
-func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan []*edge_ctrl_pb.ApiSession, syncFail chan error, timeout time.Duration) {
+func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan *apiSessionSyncTracker, syncFail chan error, timeout time.Duration) {
 	tracker.deadline.Do(func() {
 		go func() {
 			ticker := time.NewTicker(1 * time.Second)
@@ -261,7 +268,8 @@ func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan []*edge_ctrl_
 				return
 			case <-ticker.C:
 				if tracker.HasAll() {
-					syncReady <- tracker.all()
+					syncReady <- tracker
+
 					tracker.isDone = true
 					return
 				}
@@ -312,6 +320,11 @@ type apiSessionAddedWithState struct {
 	*edge_ctrl_pb.ApiSessionAdded
 }
 
+type apiSessionSync struct {
+	syncId     string
+	apiSession []*edge_ctrl_pb.ApiSession
+}
+
 func parseInstantSyncHeaders(msg *channel2.Message) (string, *sync_strats.InstantSyncState, error) {
 	if syncStrategyType, ok := msg.Headers[env.SyncStrategyTypeHeader]; ok {
 		if syncStrategyState, ok := msg.Headers[env.SyncStrategyStateHeader]; ok {

router/internal/fabric/manager.go

@@ -58,7 +58,7 @@ type StateManager interface {
 	AddApiSession(apiSession *edge_ctrl_pb.ApiSession)
 	UpdateApiSession(apiSession *edge_ctrl_pb.ApiSession)
 	RemoveApiSession(token string)
-	RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession)
+	RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession, beforeSessionId string)
 	AddConnectedApiSession(token string, removeCB func(), ch channel2.Channel)
 	RemoveConnectedApiSession(token string, underlay channel2.Channel)
 	AddApiSessionRemovedListener(token string, callBack func(token string)) RemoveListener
@@ -67,8 +67,8 @@ type StateManager interface {
 }
 
 type StateManagerImpl struct {
-	sessionsByToken    *sync.Map //"network" sessions
-	apiSessionsByToken *sync.Map
+	sessionsByToken    *sync.Map //token -> "network" sessions
+	apiSessionsByToken *sync.Map // token -> api session
 	activeApiSessions  cmap.ConcurrentMap
 
 	Hostname       string
@@ -141,16 +141,22 @@ func (sm *StateManagerImpl) RemoveApiSession(token string) {
 		pfxlog.Logger().WithField("apiSessionToken", token).Debugf("could not remove api session [token: %s]; not found", token)
 	}
 }
-func (sm *StateManagerImpl) RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession) {
+
+// Removes API Sessions not present in the knownApiSessions argument. If the beforeSessionId value is not empty string,
+// it will be used as a monotonic comparison between it and  API session ids. API session ids later than the sync
+// will be ignored.
+func (sm *StateManagerImpl) RemoveMissingApiSessions(knownApiSessions []*edge_ctrl_pb.ApiSession, beforeSessionId string) {
 	validTokens := map[string]bool{}
-	for _, session := range knownSessions {
-		validTokens[session.Token] = true
+	for _, apiSession := range knownApiSessions {
+		validTokens[apiSession.Token] = true
 	}
 
 	var tokensToRemove []string
-	sm.apiSessionsByToken.Range(func(key, _ interface{}) bool {
+	sm.apiSessionsByToken.Range(func(key, val interface{}) bool {
 		token, _ := key.(string)
-		if _, ok := validTokens[token]; !ok {
+		apiSession, _ := val.(*edge_ctrl_pb.ApiSession)
+
+		if _, ok := validTokens[token]; !ok && (beforeSessionId == "" || apiSession.Id <= beforeSessionId) {
 			tokensToRemove = append(tokensToRemove, token)
 		}
 		return true

tests/service_test.go

@@ -168,7 +168,7 @@ func Test_Services(t *testing.T) {
 		entityJson := ctx.AdminSession.validateEntityWithQuery(service)
 		createdAt := ctx.validateDateFieldsForCreate(now, entityJson)
 
-		time.Sleep(time.Millisecond * 10)
+		time.Sleep(time.Millisecond * 100)
 		now = time.Now()
 		service.terminatorStrategy = "ha"
 		ctx.AdminSession.requireUpdateEntity(service)