fix reconcile assignment

pkg/floatingip/ip_controller.go

@@ -474,14 +474,18 @@ func (i *ipController) reconcileIPStatus(ctx context.Context) {
 					nodeProviderID: "", // reset
 				}
 			}
-			i.providerIDToIP[providerID] = ip
+			if providerID != "" {
+				i.providerIDToIP[providerID] = ip
+			}
 
 			delete(i.providerIDToIP, expectedProviderID)
-			if evictedNodeName, ok := i.providerIDToNodeName[providerID]; ok {
+			// If the node we expected to hold this IP is still active, it lost its IP (either
+			// removed externally, or claimed by another node) and needs a new one.
+			if evictedNodeName, ok := i.providerIDToNodeName[expectedProviderID]; ok {
 				log.WithFields(logrus.Fields{
 					"node": evictedNodeName,
-					"ip":   expectedIP,
-				}).Info("nodes ip was claimed by other node; marking for reassignment")
+					"ip":   ip,
+				}).Info("nodes ip was removed or claimed by other node; marking node for reassignment")
 				i.assignableNodes.Add(expectedProviderID, true)
 			}
 		}
@@ -531,9 +535,10 @@ func (i *ipController) reconcileAssignment(ctx context.Context) {
 		ip := i.assignableIPs.Front()
 
 		// If this IP was previously involved in an error we shouldn't attempt to try again before
-		// its retry timestamp.
+		// its retry timestamp. Healthy IPs also carry a nextRetry timestamp (their periodic
+		// status check), which must not delay assignment.
 		status := i.ipToStatus[ip]
-		if !status.nextRetry.IsZero() && !status.nextRetry.After(now) {
+		if status.state == flipopv1alpha1.IPStateError && status.nextRetry.After(now) {
 			retryIPs = append(retryIPs, ip)
 			i.retry(status.nextRetry)
 			continue
@@ -544,7 +549,7 @@ func (i *ipController) reconcileAssignment(ctx context.Context) {
 		// Similarly, if this node was involved in an error we should wait until after its retry
 		// timestamp has elapsed.
 		nRetry, ok := i.providerIDToRetry[providerID]
-		if ok && !nRetry.nextRetry.IsZero() && !nRetry.nextRetry.After(now) {
+		if ok && nRetry.nextRetry.After(now) {
 			retryIPs = append(retryIPs, ip)
 			retryProviders = append(retryProviders, providerID)
 			i.retry(nRetry.nextRetry)
@@ -597,6 +602,12 @@ func (i *ipController) reconcileAssignment(ctx context.Context) {
 			nRetry.attempts, nRetry.nextRetry = nRetry.retrySchedule.Next(nRetry.attempts)
 			i.providerIDToRetry[providerID] = nRetry
 			i.retry(nRetry.nextRetry)
+			// The assignment failed. Roll back the speculative claim recorded above, and requeue
+			// both the IP and the node so assignment is retried once their backoff elapses.
+			status.nodeProviderID = ""
+			delete(i.providerIDToIP, providerID)
+			retryIPs = append(retryIPs, ip)
+			retryProviders = append(retryProviders, providerID)
 		}
 
 		i.dnsDirty = true
@@ -731,7 +742,6 @@ func (i *ipController) EnableNodes(ctx context.Context, nodes ...*corev1.Node) {
 			i.assignableIPs.Delete(ip)
 			// Since the IP address is already assigned to the node we want to ensure the annotation reflects it as well.
 			if i.onAnnotate != nil {
-				log.Warn("test")
 				if err := i.onAnnotate(ctx, i.providerIDToNodeName[providerID], ip); err != nil {
 					i.log.WithError(err).Error("updating Reserved IP annotation")
 				}

pkg/floatingip/ip_controller_test.go

@@ -30,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/uuid"
 
+	flipopv1alpha1 "github.com/digitalocean/flipop/pkg/apis/flipop/v1alpha1"
 	"github.com/digitalocean/flipop/pkg/log"
 	"github.com/digitalocean/flipop/pkg/provider"
 )
@@ -204,6 +205,21 @@ func TestIPControllerReconcileIPStatus(t *testing.T) {
 			expectAssignableIPs:   []string{},
 			expectAssignableNodes: []string{"mock://1"},
 		},
+		{
+			name:      "ip unassigned externally; node requeued",
+			ips:       []string{"192.168.1.1"},
+			responses: []ipToProviderIDRes{{ip: "192.168.1.1", providerID: ""}},
+			setup: func(i *ipController) {
+				i.providerIDToIP["mock://1"] = "192.168.1.1"
+				i.ipToStatus["192.168.1.1"] = &ipStatus{
+					nodeProviderID: "mock://1",
+				}
+				i.providerIDToNodeName["mock://1"] = "some-node"
+			},
+			expectProviderIDToIP:  map[string]string{},
+			expectAssignableIPs:   []string{"192.168.1.1"},
+			expectAssignableNodes: []string{"mock://1"},
+		},
 		{
 			name: "provider reports ip reassigned",
 			ips:  []string{"192.168.1.1", "172.16.2.2"},
@@ -369,18 +385,111 @@ func TestIPControllerReconcileAssignment(t *testing.T) {
 			},
 		},
 		{
-			name:                 "assignment error",
+			name:            "assignment error",
+			assignableIPs:   []string{"192.168.1.1"},
+			assignableNodes: []string{"mock://1"},
+			// The failed assignment must not leave a speculative claim behind, and both the
+			// IP and node must be requeued so assignment is retried after backoff.
+			expectProviderIDToIP:  map[string]string{},
+			responses:             []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1", err: errors.New("nope")}},
+			expectIPRetry:         true, // We always retry, because of assign
+			expectAssignableIPs:   []string{"192.168.1.1"},
+			expectAssignableNodes: []string{"mock://1"},
+			setup: func(i *ipController) {
+				i.ipToStatus["192.168.1.1"] = &ipStatus{}
+			},
+			eval: func(i *ipController) {
+				require.Equal(t, provider.RetrySlow, i.ipToStatus["192.168.1.1"].retrySchedule)
+				require.Contains(t, i.providerIDToRetry, "mock://1")
+				require.Empty(t, i.ipToStatus["192.168.1.1"].nodeProviderID)
+			},
+		},
+		{
+			name:                  "node in retry backoff is not attempted",
+			assignableIPs:         []string{"192.168.1.1"},
+			assignableNodes:       []string{"mock://1"},
+			expectProviderIDToIP:  map[string]string{},
+			responses:             nil, // AssignIP must NOT be called.
+			expectIPRetry:         true,
+			expectAssignableIPs:   []string{"192.168.1.1"},
+			expectAssignableNodes: []string{"mock://1"},
+			setup: func(i *ipController) {
+				i.ipToStatus["192.168.1.1"] = &ipStatus{}
+				i.now = func() time.Time { return fakeNow }
+				i.providerIDToRetry["mock://1"] = &retry{
+					attempts:      1,
+					nextRetry:     fakeNow.Add(30 * time.Second),
+					retrySchedule: provider.RetryFast,
+				}
+			},
+			eval: func(i *ipController) {
+				// The reconciler must wake when the node's backoff elapses.
+				require.Equal(t, fakeNow.Add(30*time.Second), i.nextRetry)
+			},
+		},
+		{
+			name:                 "node retry deadline elapsed; assignment attempted",
 			assignableIPs:        []string{"192.168.1.1"},
 			assignableNodes:      []string{"mock://1"},
 			expectProviderIDToIP: map[string]string{"mock://1": "192.168.1.1"},
-			responses:            []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1", err: errors.New("nope")}},
-			expectIPRetry:        true, // We always retry, because of assign
+			responses:            []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1"}},
+			expectIPRetry:        true,
+			expectAnnotateCall:   &annotateCall{NodeName: "hello-world", IP: "192.168.1.1"},
 			setup: func(i *ipController) {
 				i.ipToStatus["192.168.1.1"] = &ipStatus{}
+				i.providerIDToNodeName["mock://1"] = "hello-world"
+				i.now = func() time.Time { return fakeNow }
+				i.providerIDToRetry["mock://1"] = &retry{
+					attempts:      1,
+					nextRetry:     fakeNow.Add(-time.Second),
+					retrySchedule: provider.RetryFast,
+				}
 			},
 			eval: func(i *ipController) {
-				require.Equal(t, provider.RetrySlow, i.ipToStatus["192.168.1.1"].retrySchedule)
-				require.Contains(t, i.providerIDToRetry, "mock://1")
+				require.NotContains(t, i.providerIDToRetry, "mock://1")
+			},
+		},
+		{
+			name:                  "ip in error backoff is not attempted",
+			assignableIPs:         []string{"192.168.1.1"},
+			assignableNodes:       []string{"mock://1"},
+			expectProviderIDToIP:  map[string]string{},
+			responses:             nil, // AssignIP must NOT be called.
+			expectIPRetry:         true,
+			expectAssignableIPs:   []string{"192.168.1.1"},
+			expectAssignableNodes: []string{"mock://1"},
+			setup: func(i *ipController) {
+				i.now = func() time.Time { return fakeNow }
+				i.ipToStatus["192.168.1.1"] = &ipStatus{
+					state: flipopv1alpha1.IPStateError,
+					retry: retry{
+						attempts:      1,
+						nextRetry:     fakeNow.Add(30 * time.Second),
+						retrySchedule: provider.RetryFast,
+					},
+				}
+			},
+		},
+		{
+			name:                 "healthy ip with future status check is assigned immediately",
+			assignableIPs:        []string{"192.168.1.1"},
+			assignableNodes:      []string{"mock://1"},
+			expectProviderIDToIP: map[string]string{"mock://1": "192.168.1.1"},
+			responses:            []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1"}},
+			expectIPRetry:        true,
+			expectAnnotateCall:   &annotateCall{NodeName: "hello-world", IP: "192.168.1.1"},
+			setup: func(i *ipController) {
+				i.providerIDToNodeName["mock://1"] = "hello-world"
+				i.now = func() time.Time { return fakeNow }
+				// Healthy IPs carry a nextRetry for their periodic status check; this must
+				// not delay assignment.
+				i.ipToStatus["192.168.1.1"] = &ipStatus{
+					state: flipopv1alpha1.IPStateUnassigned,
+					retry: retry{
+						nextRetry:     fakeNow.Add(5 * time.Minute),
+						retrySchedule: healthyRetrySchedule,
+					},
+				}
 			},
 		},
 	}

pkg/provider/digitalocean.go

@@ -209,6 +209,11 @@ func (do *digitalOcean) AssignIP(ctx context.Context, ip, providerID string) (er
 						return ErrNodeInUse
 					}
 				}
+				if isPendingEventError(err) {
+					// The droplet has another event in-flight (common during node rotation
+					// or provisioning). These resolve quickly; retry on the fast schedule.
+					return NewRetryError(err, RetryFast)
+				}
 			}
 		}
 		return err
@@ -290,6 +295,12 @@ func (do *digitalOcean) NodeToIP(ctx context.Context, providerID string) (_ stri
 	return "", nil
 }
 
+// isPendingEventError returns true if the error indicates the droplet already has an event
+// in-flight (API status 422, "Droplet already has a pending event").
+func isPendingEventError(err error) bool {
+	return err != nil && strings.Contains(strings.ToLower(err.Error()), "pending event")
+}
+
 // asyncStatus tries to abstract the asynchronous nature of DO's floating IP updates.
 func (do *digitalOcean) asyncStatus(ctx context.Context, ip string) error {
 	do.lock.Lock()

pkg/provider/digitalocean_test.go

@@ -19,6 +19,8 @@ package provider
 import (
 	"context"
 	"net"
+	"net/http"
+	"net/url"
 	"testing"
 
 	"github.com/digitalocean/godo"
@@ -31,6 +33,69 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestDigitalOceanAssignIP422Classification(t *testing.T) {
+	tcs := []struct {
+		name           string
+		message        string
+		expectSchedule RetrySchedule
+	}{
+		{
+			// A droplet with an event in-flight (e.g. during node rotation) rejects
+			// assignment with a 422. This resolves quickly and must be retried fast.
+			name:           "pending event retries fast",
+			message:        "Droplet already has a pending event.",
+			expectSchedule: RetryFast,
+		},
+		{
+			name:           "other 422 retries slow",
+			message:        "cannot assign a reserved IP to a droplet in a different region",
+			expectSchedule: RetrySlow,
+		},
+	}
+	for _, tc := range tcs {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+
+			httpRes := &http.Response{
+				StatusCode: http.StatusUnprocessableEntity,
+				Request: &http.Request{
+					Method: http.MethodPost,
+					URL:    &url.URL{},
+				},
+			}
+			assignErr := &godo.ErrorResponse{
+				Response: httpRes,
+				Message:  tc.message,
+			}
+
+			ipActionsService := mock_godo.NewMockFloatingIPActionsService(ctrl)
+			ipActionsService.EXPECT().
+				Assign(gomock.Any(), "10.0.0.1", 12345).
+				Return(nil, &godo.Response{Response: httpRes}, assignErr)
+
+			// AssignIP consults the IP's current assignment before classifying the 422.
+			// Report it unassigned so classification falls through to the error itself.
+			ipsService := mock_godo.NewMockFloatingIPsService(ctrl)
+			ipsService.EXPECT().
+				Get(gomock.Any(), "10.0.0.1").
+				Return(&godo.FloatingIP{}, &godo.Response{}, nil)
+
+			do := &digitalOcean{
+				ipsService:        ipsService,
+				ipActionsService:  ipActionsService,
+				floatingIPActions: make(map[string]*doAction),
+				log:               log.NewTestLogger(t),
+			}
+			err := do.AssignIP(ctx, "10.0.0.1", "digitalocean://12345")
+			require.Error(t, err)
+			require.Equal(t, tc.expectSchedule, ErrorToRetrySchedule(err))
+		})
+	}
+}
+
 func TestDigitalOceanEnsureDNSARecordSet(t *testing.T) {
 	tcs := []struct {
 		name         string