feat(cns): make unified DEL transactional

cns/restserver/durable_state_adapter.go

@@ -61,6 +61,26 @@ type durableStateOperations struct {
 		time.Duration,
 		func(state.Snapshot) error,
 	) (bool, error)
+	releaseEndpoint func(
+		context.Context,
+		uint64,
+		state.PodIdentity,
+		time.Time,
+		func(state.Snapshot) error,
+	) (bool, error)
+	deleteEndpoint func(
+		context.Context,
+		uint64,
+		string,
+		func(state.Snapshot) error,
+	) (bool, error)
+	pruneDeleteIntents func(
+		context.Context,
+		uint64,
+		time.Time,
+		time.Duration,
+		func(state.Snapshot) error,
+	) (int, error)
 	refreshMetrics func(context.Context) (state.Status, error)
 	status         func(context.Context) (state.Status, error)
 	close          func() error
@@ -72,15 +92,16 @@ type durableStateAdapter struct {
 
 	// mu is acquired before the HTTPRestService lock. Callers must not hold the
 	// service lock; the adapter applies complete projections under that lock.
-	mu                   sync.Mutex
-	projectEndpointState bool
-	buildProjection      func(state.Snapshot) (durableCacheProjection, error)
-	applyAddProjection   func(durableCacheProjection) error
-	now                  func() time.Time
-	projected            bool
-	generation           uint64
-	closeOnce            sync.Once
-	closeErr             error
+	mu                    sync.Mutex
+	projectEndpointState  bool
+	buildProjection       func(state.Snapshot) (durableCacheProjection, error)
+	applyAddProjection    func(durableCacheProjection) error
+	applyDeleteProjection func(durableCacheProjection) error
+	now                   func() time.Time
+	projected             bool
+	generation            uint64
+	closeOnce             sync.Once
+	closeErr              error
 }
 
 type durableServiceMetadata struct {
@@ -137,10 +158,13 @@ func newDurableStateAdapter(
 		return nil, errNilDurableStateDB
 	}
 	return newDurableStateAdapterWithOperations(service, durableStateOperations{
-		snapshot:       db.Snapshot,
-		replace:        db.ReplaceDurableState,
-		assignEndpoint: db.AssignEndpointIfGeneration,
-		refreshMetrics: db.RefreshMetrics,
+		snapshot:           db.Snapshot,
+		replace:            db.ReplaceDurableState,
+		assignEndpoint:     db.AssignEndpointIfGeneration,
+		releaseEndpoint:    db.ReleaseEndpointIfGeneration,
+		deleteEndpoint:     db.DeleteEndpointRecordIfGeneration,
+		pruneDeleteIntents: db.PruneDeleteIntentsIfGeneration,
+		refreshMetrics:     db.RefreshMetrics,
 		updateMetadata: func(ctx context.Context, expectedGeneration uint64, metadata state.Metadata) (bool, error) {
 			err := db.Update(ctx, func(tx *state.WriteTx) error {
 				current, err := tx.Metadata()

cns/restserver/ipam.go

@@ -454,8 +454,18 @@ func (service *HTTPRestService) ReleaseIPConfigHandlerHelper(ctx context.Context
 			},
 		}, fmt.Errorf("failed to validate ip config request") //nolint:goerr113 // return error
 	}
-	// Check if http rest service managed endpoint state is set
-	if service.Options[common.OptManageEndpointState] == true {
+	unifiedAdapter := service.selectedUnifiedStateAdapter()
+	if unifiedAdapter != nil {
+		if err := unifiedAdapter.releaseIPConfigs(ctx, ipconfigsRequest, podInfo); err != nil {
+			resp := &cns.IPConfigsResponse{
+				Response: cns.Response{
+					ReturnCode: unifiedReleaseResponseCode(err),
+					Message:    err.Error(),
+				},
+			}
+			return resp, fmt.Errorf("releasing unified IP configs: %w", err)
+		}
+	} else if service.Options[common.OptManageEndpointState] == true {
 		if err := service.releaseIPConfigsWithDeleteIntent(podInfo); err != nil {
 			resp := &cns.IPConfigsResponse{
 				Response: cns.Response{
@@ -1221,6 +1231,12 @@ func validateDesiredIPAddresses(desiredIPs []string) error {
 func (service *HTTPRestService) EndpointHandlerAPI(w http.ResponseWriter, r *http.Request) {
 	opName := "endpointHandler"
 	logger.Printf("[EndpointHandlerAPI] EndpointHandlerAPI received request with http Method %s", r.Method)
+	if r.Method == http.MethodDelete {
+		if adapter := service.selectedUnifiedStateAdapter(); adapter != nil {
+			service.deleteEndpointStateHandler(w, r, adapter)
+			return
+		}
+	}
 	service.Lock()
 	defer service.Unlock()
 	// Check if CNS is managing the CNI statefile
@@ -1239,19 +1255,27 @@ func (service *HTTPRestService) EndpointHandlerAPI(w http.ResponseWriter, r *htt
 	case http.MethodPatch:
 		service.UpdateEndpointHandler(w, r)
 	case http.MethodDelete:
-		service.DeleteEndpointStateHandler(w, r)
+		service.deleteEndpointStateHandler(w, r, nil)
 	default:
 		//nolint
 		logger.Errorf("[EndpointHandlerAPI] EndpointHandler API expect http Get or Patch or Delete method")
 	}
 }
 
 func (service *HTTPRestService) DeleteEndpointStateHandler(w http.ResponseWriter, r *http.Request) {
+	service.deleteEndpointStateHandler(w, r, service.selectedUnifiedStateAdapter())
+}
+
+func (service *HTTPRestService) deleteEndpointStateHandler(
+	w http.ResponseWriter,
+	r *http.Request,
+	adapter *durableStateAdapter,
+) {
 	opName := "DeleteEndpointStateHandler"
 	logger.Printf("[DeleteEndpointStateHandler] DeleteEndpointState for %s", r.URL.Path) //nolint:staticcheck // reason: using deprecated call until migration to new API
 	endpointID := strings.TrimPrefix(r.URL.Path, cns.EndpointPath)
 
-	if service.EndpointStateStore == nil {
+	if service.EndpointStateStore == nil && adapter == nil {
 		response := cns.Response{
 			ReturnCode: types.NilEndpointStateStore,
 			Message:    "[DeleteEndpointStateHandler] EndpointStateStore is not initialized",
@@ -1262,7 +1286,7 @@ func (service *HTTPRestService) DeleteEndpointStateHandler(w http.ResponseWriter
 	}
 
 	// Delete the endpoint from state
-	err := service.DeleteEndpointStateHelper(endpointID)
+	err := service.deleteEndpointState(r.Context(), endpointID, adapter)
 	if err != nil {
 		response := cns.Response{
 			ReturnCode: types.UnexpectedError,
@@ -1310,6 +1334,17 @@ func (service *HTTPRestService) DeleteEndpointStateHelper(endpointID string) err
 	return nil
 }
 
+func (service *HTTPRestService) deleteEndpointState(
+	ctx context.Context,
+	endpointID string,
+	adapter *durableStateAdapter,
+) error {
+	if adapter != nil {
+		return adapter.deleteEndpointRecord(ctx, endpointID)
+	}
+	return service.DeleteEndpointStateHelper(endpointID)
+}
+
 // GetEndpointHandler handles the incoming GetEndpoint requests with http Get method
 func (service *HTTPRestService) GetEndpointHandler(w http.ResponseWriter, r *http.Request) {
 	opName := "getEndpointState"

cns/restserver/unified_add.go

@@ -79,7 +79,8 @@ func (a *durableStateAdapter) requestIPConfigs(
 	if err != nil {
 		return nil, err
 	}
-	plan, err := a.service.requestIPConfigsUnifiedLocked(ctx, request, podInfo, snapshot)
+	now := a.now()
+	plan, err := a.service.requestIPConfigsUnifiedLocked(ctx, request, podInfo, snapshot, now)
 	if err != nil {
 		return nil, err
 	}
@@ -93,7 +94,7 @@ func (a *durableStateAdapter) requestIPConfigs(
 		a.generation,
 		plan.assignment,
 		plan.endpoint,
-		a.now(),
+		now,
 		unifiedDeleteIntentTTL,
 		func(candidate state.Snapshot) error {
 			var buildErr error
@@ -158,6 +159,7 @@ func (service *HTTPRestService) requestIPConfigsUnifiedLocked(
 	request cns.IPConfigsRequest,
 	podInfo cns.PodInfo,
 	snapshot state.Snapshot,
+	now time.Time,
 ) (unifiedAddPlan, error) {
 	if err := ctx.Err(); err != nil {
 		return unifiedAddPlan{}, fmt.Errorf("planning unified endpoint assignment: %w", err)
@@ -169,6 +171,14 @@ func (service *HTTPRestService) requestIPConfigsUnifiedLocked(
 		PodName:          podInfo.Name(),
 		PodNamespace:     podInfo.Namespace(),
 	}
+	if intent, ok := snapshot.DeleteIntents[requestedPod.InfraContainerID]; ok &&
+		now.Before(intent.CreatedAt.Add(unifiedDeleteIntentTTL)) {
+		return unifiedAddPlan{}, fmt.Errorf(
+			"%w: infra container %q",
+			state.ErrDeleteIntent,
+			requestedPod.InfraContainerID,
+		)
+	}
 	if existing, ok := snapshot.Assignments[requestedPod.PodKey]; ok {
 		if existing.Pod != requestedPod {
 			return unifiedAddPlan{}, fmt.Errorf(