diff --git a/pkg/epp/controller/pod_reconciler.go b/pkg/epp/controller/pod_reconciler.go
index 2d0d32c049..ea2065fbfb 100644
--- a/pkg/epp/controller/pod_reconciler.go
+++ b/pkg/epp/controller/pod_reconciler.go
@@ -59,7 +59,9 @@ func (c *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 		return ctrl.Result{}, fmt.Errorf("unable to get pod - %w", err)
 	}
 
-	c.updateDatastore(ctx, pod)
+	if err := c.updateDatastore(ctx, pod); err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to update datastore for pod %s - %w", req.NamespacedName, err)
+	}
 	return ctrl.Result{}, nil
 }
 
@@ -91,16 +93,12 @@ func (c *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(c)
 }
 
-func (c *PodReconciler) updateDatastore(ctx context.Context, pod *corev1.Pod) {
+func (c *PodReconciler) updateDatastore(ctx context.Context, pod *corev1.Pod) error {
 	logger := log.FromContext(ctx)
 	if !podutil.IsPodReady(pod) || !c.Datastore.PoolLabelsMatch(pod.Labels) {
 		logger.V(logutil.DEBUG).Info("Pod removed or not added")
 		c.Datastore.PodDelete(pod.Name)
-	} else {
-		if c.Datastore.PodUpdateOrAddIfNotExist(ctx, pod) {
-			logger.V(logutil.DEFAULT).Info("Pod added")
-		} else {
-			logger.V(logutil.DEFAULT).Info("Pod already exists")
-		}
+		return nil
 	}
+	return c.Datastore.PodUpdateOrAddIfNotExist(ctx, pod)
 }
diff --git a/pkg/epp/controller/pod_reconciler_test.go b/pkg/epp/controller/pod_reconciler_test.go
index a8ab97556e..ae62d6be75 100644
--- a/pkg/epp/controller/pod_reconciler_test.go
+++ b/pkg/epp/controller/pod_reconciler_test.go
@@ -35,6 +35,7 @@ import (
 
 	"github.com/llm-d/llm-d-router/pkg/epp/datalayer"
 	"github.com/llm-d/llm-d-router/pkg/epp/datastore"
+	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
 	"github.com/llm-d/llm-d-router/pkg/epp/util/pool"
 	testutil "github.com/llm-d/llm-d-router/pkg/epp/util/testing"
 )
@@ -202,7 +203,7 @@ func TestPodReconciler(t *testing.T) {
 				store := datastore.NewDatastore(t.Context(), epf)
 				_ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(test.pool))
 				for _, pod := range test.existingPods {
-					store.PodUpdateOrAddIfNotExist(t.Context(), pod)
+					_ = store.PodUpdateOrAddIfNotExist(t.Context(), pod)
 				}
 
 				podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store}
@@ -229,3 +230,40 @@ func TestPodReconciler(t *testing.T) {
 		}
 	}
 }
+
+// TestPodReconciler_ErrorsOnRegistrationDrop verifies that Reconcile surfaces a dropped endpoint
+// registration as an error so controller-runtime requeues the pod (#2060).
+func TestPodReconciler_ErrorsOnRegistrationDrop(t *testing.T) {
+	scheme := runtime.NewScheme()
+	_ = clientgoscheme.AddToScheme(scheme)
+
+	incomingPod := testutil.FromBase(basePod1).
+		Labels(map[string]string{"some-key": "some-val"}).
+		ReadyCondition().ObjRef()
+	fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(incomingPod).Build()
+
+	testPool := &v1.InferencePool{
+		Spec: v1.InferencePoolSpec{
+			TargetPorts: []v1.Port{{Number: v1.PortNumber(int32(8000))}},
+			Selector: v1.LabelSelector{
+				MatchLabels: map[v1.LabelKey]v1.LabelValue{"some-key": "some-val"},
+			},
+		},
+	}
+	// The factory stands in for a collector that is still registered for the endpoint (upsert
+	// overlapping an in-flight delete) or fails to start.
+	store := datastore.NewDatastore(t.Context(), &datalayer.FakeEndpointFactory{
+		NewEndpointFn: func(_ context.Context, _ *fwkdl.EndpointMetadata) fwkdl.Endpoint { return nil },
+	})
+	_ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(testPool))
+
+	podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store}
+	req := ctrl.Request{NamespacedName: types.NamespacedName{Name: incomingPod.Name, Namespace: incomingPod.Namespace}}
+
+	if _, err := podReconciler.Reconcile(context.Background(), req); err == nil {
+		t.Error("expected Reconcile to return an error for a dropped endpoint registration, got nil")
+	}
+	if pods := store.PodList(datastore.AllPodsPredicate); len(pods) != 0 {
+		t.Errorf("expected no pods in datastore, got %d", len(pods))
+	}
+}
diff --git a/pkg/epp/datalayer/fake_factory.go b/pkg/epp/datalayer/fake_factory.go
new file mode 100644
index 0000000000..dde55d538d
--- /dev/null
+++ b/pkg/epp/datalayer/fake_factory.go
@@ -0,0 +1,53 @@
+/*
+Copyright 2026 The llm-d Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package datalayer
+
+import (
+	"context"
+
+	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
+)
+
+// FakeEndpointFactory is a configurable EndpointFactory for tests. Each hook overrides the
+// corresponding method. A nil NewEndpointFn creates a plain endpoint with fresh metrics; nil
+// UpdateEndpointFn and ReleaseEndpointFn hooks are no-ops.
+type FakeEndpointFactory struct {
+	NewEndpointFn     func(ctx context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint
+	UpdateEndpointFn  func(ctx context.Context, ep fwkdl.Endpoint)
+	ReleaseEndpointFn func(ep fwkdl.Endpoint)
+}
+
+var _ EndpointFactory = (*FakeEndpointFactory)(nil)
+
+func (f *FakeEndpointFactory) NewEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint {
+	if f.NewEndpointFn != nil {
+		return f.NewEndpointFn(ctx, meta)
+	}
+	return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics())
+}
+
+func (f *FakeEndpointFactory) UpdateEndpoint(ctx context.Context, ep fwkdl.Endpoint) {
+	if f.UpdateEndpointFn != nil {
+		f.UpdateEndpointFn(ctx, ep)
+	}
+}
+
+func (f *FakeEndpointFactory) ReleaseEndpoint(ep fwkdl.Endpoint) {
+	if f.ReleaseEndpointFn != nil {
+		f.ReleaseEndpointFn(ep)
+	}
+}
diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go
index b7260474a4..eaba703b58 100644
--- a/pkg/epp/datastore/datastore.go
+++ b/pkg/epp/datastore/datastore.go
@@ -44,7 +44,11 @@ import (
 
 var (
 	errPoolNotSynced = errors.New("InferencePool is not initialized in data store")
-	AllPodsPredicate = func(_ fwkdl.Endpoint) bool { return true }
+	// errRegistrationDropped reports an endpoint that could not be tracked: its collector is
+	// still registered from an earlier registration (an upsert overlapping an in-flight delete)
+	// or failed to start. Callers match it with errors.Is to decide whether to retry.
+	errRegistrationDropped = errors.New("endpoint registration dropped: collector already registered or failed to start")
+	AllPodsPredicate       = func(_ fwkdl.Endpoint) bool { return true }
 )
 
 const (
@@ -88,10 +92,15 @@ type Datastore interface {
 
 	// PodList lists pods matching the given predicate.
 	PodList(predicate func(fwkdl.Endpoint) bool) []fwkdl.Endpoint
-	PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) bool
+	// PodUpdateOrAddIfNotExist stores or updates the endpoints for the given pod. It returns an
+	// error when an endpoint registration was dropped (see upsertEndpoint); the pod is then not
+	// tracked by the datastore and the caller must retry (e.g. by requeuing the reconcile).
+	PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) error
 	PodDelete(podName string)
 
 	// EndpointUpsert adds or updates an endpoint from a non-Kubernetes discovery source.
+	// A dropped registration is logged; the endpoint stays untracked until the discovery
+	// source re-emits it.
 	EndpointUpsert(ctx context.Context, meta *fwkdl.EndpointMetadata)
 	// EndpointDelete removes the endpoint with the given namespaced name.
 	EndpointDelete(id types.NamespacedName)
@@ -130,6 +139,11 @@ type datastore struct {
 	// key: types.NamespacedName, value: fwkdl.Endpoint
 	pods *sync.Map
 	epf  datalayer.EndpointFactory
+	// needsResync forces the next PoolSet to run podResyncAll even when the pool is unchanged.
+	// PoolSet stores the pool before resyncing, so without this flag a PoolSet retried after a
+	// resync failure would compare the incoming pool against the already-stored identical pool
+	// and skip the resync. Guarded by mu.
+	needsResync bool
 }
 
 func (ds *datastore) WithEndpointPool(pool *datalayer.EndpointPool) Datastore {
@@ -167,7 +181,7 @@ func (ds *datastore) PoolSet(ctx context.Context, reader client.Reader, endpoint
 	selectorChanged := oldEndpointPool == nil || !selectorEqual(oldEndpointPool.Selector, endpointPool.Selector)
 	targetPortsChanged := oldEndpointPool != nil && !slices.Equal(oldEndpointPool.TargetPorts, endpointPool.TargetPorts)
 
-	if selectorChanged || targetPortsChanged {
+	if selectorChanged || targetPortsChanged || ds.needsResync {
 		logger.V(logutil.DEFAULT).Info("Updating endpoints", "selector", endpointPool.Selector, "targetPortsChanged", targetPortsChanged)
 		// A full resync is required to address the following cases:
 		// 1) At startup, the pod events may get processed before the pool is synced with the datastore,
@@ -178,8 +192,10 @@ func (ds *datastore) PoolSet(ctx context.Context, reader client.Reader, endpoint
 		// 3) If the targetPorts changed, we need to resync to remove orphaned rank endpoints that no longer
 		//    exist in the new targetPorts configuration.
 		if err := ds.podResyncAll(ctx, reader); err != nil {
+			ds.needsResync = true
 			return fmt.Errorf("failed to update pods according to the pool selector - %w", err)
 		}
+		ds.needsResync = false
 	}
 
 	return nil
@@ -282,7 +298,7 @@ func (ds *datastore) PodList(predicate func(fwkdl.Endpoint) bool) []fwkdl.Endpoi
 	return res
 }
 
-func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) bool {
+func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) error {
 	// Take a reference to pool under read lock to avoid racing with PoolSet().
 	// This is safe because PoolSet() replaces the entire pool struct rather than
 	// updating it in-place.
@@ -290,15 +306,22 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P
 	pool := ds.pool
 	ds.mu.RUnlock()
 
+	if pool == nil {
+		// Without the pool's target ports the pod cannot be mapped to endpoints; the resync
+		// triggered when the pool syncs picks the pod up.
+		log.FromContext(ctx).V(logutil.DEBUG).Info("Skipping pod upsert, InferencePool not synced", "name", pod.Name)
+		return nil
+	}
 	return ds.podUpdateOrAddIfNotExist(ctx, pod, pool)
 }
 
 // podUpdateOrAddIfNotExist is the lock-free inner implementation.
-// Callers must ensure pool is a consistent snapshot (either read under lock
+// Callers must ensure pool is a non-nil consistent snapshot (either read under lock
 // or already held, as in podResyncAll which runs under ds.mu.Lock via PoolSet).
-func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod, pool *datalayer.EndpointPool) bool {
+// It returns a joined error covering every endpoint of the pod whose registration was dropped.
+func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod, pool *datalayer.EndpointPool) error {
 	if pool == nil {
-		return true
+		return nil
 	}
 
 	labels := make(map[string]string, len(pod.GetLabels()))
@@ -328,12 +351,26 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P
 			"pod", pod.Name, "namespace", pod.Namespace, "targetPorts", pool.TargetPorts)
 	}
 
-	result := true
+	added := false
+	var errs []error
 	existingEpSet := sets.Set[types.NamespacedName]{}
 	for _, endpointMetadata := range pods {
 		existingEpSet.Insert(endpointMetadata.NamespacedName)
-		if ds.upsertEndpoint(ctx, endpointMetadata) {
-			result = false
+		created, err := ds.upsertEndpoint(ctx, endpointMetadata)
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		if created {
+			added = true
+		}
+	}
+	logger := log.FromContext(ctx)
+	if len(errs) == 0 {
+		if added {
+			logger.V(logutil.DEFAULT).Info("Pod added", "name", pod.Name)
+		} else {
+			logger.V(logutil.DEFAULT).Info("Pod already exists", "name", pod.Name)
 		}
 	}
 
@@ -350,7 +387,7 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P
 		}
 	}
 
-	return result
+	return errors.Join(errs...)
 }
 
 func (ds *datastore) PodDelete(podName string) {
@@ -365,7 +402,9 @@ func (ds *datastore) PodDelete(podName string) {
 }
 
 func (ds *datastore) EndpointUpsert(ctx context.Context, meta *fwkdl.EndpointMetadata) {
-	ds.upsertEndpoint(ctx, meta)
+	if _, err := ds.upsertEndpoint(ctx, meta); err != nil {
+		log.FromContext(ctx).Error(err, "failed to register endpoint", "endpoint", meta.NamespacedName)
+	}
 }
 
 func (ds *datastore) EndpointDelete(id types.NamespacedName) {
@@ -375,29 +414,38 @@ func (ds *datastore) EndpointDelete(id types.NamespacedName) {
 }
 
 // upsertEndpoint stores or updates a single endpoint in the pods map.
-// Returns true if the endpoint was newly created, false if it already existed
-// or if NewEndpoint returned nil (duplicate-start race).
+// Returns true if the endpoint was newly created, false if it already existed.
 // Shared by EndpointUpsert and podUpdateOrAddIfNotExist.
-func (ds *datastore) upsertEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) bool {
-	existing, ok := ds.pods.Load(meta.NamespacedName)
-	if !ok {
+//
+// It returns an error wrapping errRegistrationDropped when the endpoint cannot be tracked:
+// NewEndpoint returns nil when a collector is still registered for this endpoint or the
+// collector failed to start. When a concurrent upsert has stored an entry for the key, this
+// call's metadata is applied through the update path; when no entry exists (the upsert
+// overlapped an in-flight delete that removed the entry before deregistering the collector,
+// or the collector failed to start), the endpoint is untracked and the caller must retry.
+func (ds *datastore) upsertEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) (bool, error) {
+	for {
+		if existing, ok := ds.pods.Load(meta.NamespacedName); ok {
+			ep := existing.(fwkdl.Endpoint)
+			if ep.GetMetadata().Equal(meta) {
+				return false, nil
+			}
+			ep.UpdateMetadata(meta)
+			ds.epf.UpdateEndpoint(ctx, ep)
+			return false, nil
+		}
 		ep := ds.epf.NewEndpoint(ds.parentCtx, meta)
 		if ep == nil {
-			// NewEndpoint returns nil when a collector is already running for this
-			// endpoint (duplicate reconcile race). The existing entry in ds.pods
-			// is still valid; skip re-registering it.
-			return false
+			if _, ok := ds.pods.Load(meta.NamespacedName); ok {
+				// A concurrent upsert won the registration; apply this call's metadata through
+				// the update path above.
+				continue
+			}
+			return false, fmt.Errorf("endpoint %s: %w", meta.NamespacedName, errRegistrationDropped)
 		}
 		ds.pods.Store(meta.NamespacedName, ep)
-		return true
+		return true, nil
 	}
-	ep := existing.(fwkdl.Endpoint)
-	if ep.GetMetadata().Equal(meta) {
-		return false
-	}
-	ep.UpdateMetadata(meta)
-	ds.epf.UpdateEndpoint(ctx, ep)
-	return false
 }
 
 func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) error {
@@ -413,19 +461,18 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err
 	// Track active endpoints by their full name (including rank suffix).
 	// This ensures orphaned rank endpoints are removed when targetPorts shrinks.
 	activeEndpoints := sets.New[types.NamespacedName]()
+	var errs []error
 	for _, pod := range podList.Items {
 		if !podutil.IsPodReady(&pod) {
 			continue
 		}
-		namespacedName := types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}
 		// Calculate expected endpoint names based on current targetPorts.
 		for idx := range ds.pool.TargetPorts {
 			activeEndpoints.Insert(createEndpointNamespacedName(&pod, idx))
 		}
-		if !ds.podUpdateOrAddIfNotExist(ctx, &pod, ds.pool) {
-			logger.V(logutil.DEFAULT).Info("Pod added", "name", namespacedName)
-		} else {
-			logger.V(logutil.DEFAULT).Info("Pod already exists", "name", namespacedName)
+		if err := ds.podUpdateOrAddIfNotExist(ctx, &pod, ds.pool); err != nil {
+			// Propagate so PoolSet fails; needsResync makes the retried PoolSet resync again.
+			errs = append(errs, err)
 		}
 	}
 
@@ -441,7 +488,7 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err
 		return true
 	})
 
-	return nil
+	return errors.Join(errs...)
 }
 
 // extractActivePorts extracts the active ports from a pod's annotations.
diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go
index 7492ff31a1..73d1bd5819 100644
--- a/pkg/epp/datastore/datastore_test.go
+++ b/pkg/epp/datastore/datastore_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -422,7 +423,7 @@ func TestMetrics(t *testing.T) {
 			ds := NewDatastore(ctx, epf)
 			_ = ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool))
 			for _, pod := range test.storePods {
-				ds.PodUpdateOrAddIfNotExist(ctx, pod)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, pod)
 			}
 			time.Sleep(1 * time.Second) // Give some time for the metrics to be fetched.
 			if test.predict == nil {
@@ -455,7 +456,7 @@ func TestPods(t *testing.T) {
 			existingPods: []*corev1.Pod{},
 			wantPods:     []*corev1.Pod{pod1},
 			op: func(ctx context.Context, ds Datastore) {
-				ds.PodUpdateOrAddIfNotExist(ctx, pod1)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, pod1)
 			},
 		},
 		{
@@ -463,7 +464,7 @@ func TestPods(t *testing.T) {
 			existingPods: []*corev1.Pod{pod1},
 			wantPods:     []*corev1.Pod{pod1, pod2},
 			op: func(ctx context.Context, ds Datastore) {
-				ds.PodUpdateOrAddIfNotExist(ctx, pod2)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, pod2)
 			},
 		},
 		{
@@ -497,7 +498,7 @@ func TestPods(t *testing.T) {
 					t.Error(err)
 				}
 				for _, pod := range test.existingPods {
-					ds.PodUpdateOrAddIfNotExist(ctx, pod)
+					_ = ds.PodUpdateOrAddIfNotExist(ctx, pod)
 				}
 
 				test.op(ctx, ds)
@@ -655,7 +656,7 @@ func TestEndpointMetadata(t *testing.T) {
 				},
 			},
 			op: func(ctx context.Context, ds Datastore) {
-				ds.PodUpdateOrAddIfNotExist(ctx, pod1)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, pod1)
 			},
 			pool: inferencePool,
 		},
@@ -690,7 +691,7 @@ func TestEndpointMetadata(t *testing.T) {
 				},
 			},
 			op: func(ctx context.Context, ds Datastore) {
-				ds.PodUpdateOrAddIfNotExist(ctx, pod1)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, pod1)
 			},
 			pool: inferencePoolMultiTarget,
 		},
@@ -750,7 +751,7 @@ func TestEndpointMetadata(t *testing.T) {
 				},
 			},
 			op: func(ctx context.Context, ds Datastore) {
-				ds.PodUpdateOrAddIfNotExist(ctx, pod2)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, pod2)
 			},
 			pool: inferencePoolMultiTarget,
 		},
@@ -805,7 +806,7 @@ func TestEndpointMetadata(t *testing.T) {
 					t.Error(err)
 				}
 				for _, pod := range test.existingPods {
-					ds.PodUpdateOrAddIfNotExist(ctx, pod)
+					_ = ds.PodUpdateOrAddIfNotExist(ctx, pod)
 				}
 
 				test.op(ctx, ds)
@@ -964,7 +965,7 @@ func TestActivePortFiltering(t *testing.T) {
 
 				// Add all pods
 				for _, pod := range test.pods {
-					ds.PodUpdateOrAddIfNotExist(ctx, pod)
+					_ = ds.PodUpdateOrAddIfNotExist(ctx, pod)
 				}
 
 				// Check final endpoint count
@@ -1064,7 +1065,7 @@ func TestActivePortEndpointRemoval(t *testing.T) {
 			operations: []func(Datastore){
 				// Update the pod to reduce active ports from 3 to 1
 				func(ds Datastore) {
-					ds.PodUpdateOrAddIfNotExist(context.Background(), updatedPod1)
+					_ = ds.PodUpdateOrAddIfNotExist(context.Background(), updatedPod1)
 				},
 			},
 			wantEndpointCount: 1, // Only port 8000 should remain active
@@ -1080,7 +1081,7 @@ func TestActivePortEndpointRemoval(t *testing.T) {
 			operations: []func(Datastore){
 				// Update the pod to have no active ports
 				func(ds Datastore) {
-					ds.PodUpdateOrAddIfNotExist(context.Background(), inactivePod1)
+					_ = ds.PodUpdateOrAddIfNotExist(context.Background(), inactivePod1)
 				},
 			},
 			wantEndpointCount: 0, // No ports should remain active
@@ -1112,7 +1113,7 @@ func TestActivePortEndpointRemoval(t *testing.T) {
 				}
 
 				// Add the initial pod
-				ds.PodUpdateOrAddIfNotExist(ctx, test.initialPod)
+				_ = ds.PodUpdateOrAddIfNotExist(ctx, test.initialPod)
 
 				// Wait a bit for the datastore to process the pod
 				time.Sleep(100 * time.Millisecond)
@@ -1199,7 +1200,7 @@ func TestPodUpdateOrAddIfNotExist_ConcurrentPoolSet(t *testing.T) {
 			go func() {
 				defer wg.Done()
 				for range 1000 {
-					ds.PodUpdateOrAddIfNotExist(ctx, pod)
+					_ = ds.PodUpdateOrAddIfNotExist(ctx, pod)
 				}
 			}()
 
@@ -1454,6 +1455,135 @@ func TestEndpointDelete_Missing(t *testing.T) {
 	})
 }
 
+// TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError verifies that a dropped endpoint
+// registration (NewEndpoint returns nil with no surviving pods-map entry) is reported to the
+// caller instead of leaving the pod silently untracked, and that a retry succeeds once the
+// collector can start again (#2060).
+func TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError(t *testing.T) {
+	ctx := context.Background()
+	factory := &mockEndpointFactory{returnNil: true}
+	ds := NewDatastore(ctx, factory)
+	require.NoError(t, ds.PoolSet(ctx, fake.NewFakeClient(), poolutil.InferencePoolToEndpointPool(inferencePool)))
+
+	err := ds.PodUpdateOrAddIfNotExist(ctx, pod1)
+	require.ErrorIs(t, err, errRegistrationDropped)
+	assert.Empty(t, ds.PodList(AllPodsPredicate))
+
+	// The retry (the reconciler's requeue) succeeds once the collector can start.
+	factory.returnNil = false
+	require.NoError(t, ds.PodUpdateOrAddIfNotExist(ctx, pod1))
+	assert.Len(t, ds.PodList(AllPodsPredicate), 1)
+}
+
+// TestPoolSet_ResyncRetriedAfterDropError verifies that a PoolSet retried with an identical pool
+// after a failed resync runs the resync again rather than skipping it because the pool compares
+// equal to the one already stored (#2060).
+func TestPoolSet_ResyncRetriedAfterDropError(t *testing.T) {
+	ctx := context.Background()
+	readyPod := testutil.FromBase(pod1).ReadyCondition().ObjRef()
+	fakeClient := fake.NewClientBuilder().WithObjects(readyPod).Build()
+
+	fail := true
+	factory := &datalayer.FakeEndpointFactory{
+		NewEndpointFn: func(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint {
+			if fail {
+				return nil
+			}
+			return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics())
+		},
+	}
+	ds := NewDatastore(ctx, factory)
+
+	err := ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool))
+	require.ErrorIs(t, err, errRegistrationDropped)
+	assert.Empty(t, ds.PodList(AllPodsPredicate))
+
+	fail = false
+	require.NoError(t, ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool)))
+	assert.Len(t, ds.PodList(AllPodsPredicate), 1)
+}
+
+// TestUpsertEndpoint_ConcurrentStoreDuringNilAppliesMetadata verifies the duplicate-start race
+// is not reported as an error and does not lose this call's metadata: when a concurrent upsert
+// has stored an entry by the time NewEndpoint returns nil, the upsert applies its metadata to
+// that entry through the update path.
+func TestUpsertEndpoint_ConcurrentStoreDuringNilAppliesMetadata(t *testing.T) {
+	ctx := context.Background()
+	id := types.NamespacedName{Name: "ep1", Namespace: "default"}
+
+	var ds *datastore
+	factory := &datalayer.FakeEndpointFactory{
+		NewEndpointFn: func(_ context.Context, _ *fwkdl.EndpointMetadata) fwkdl.Endpoint {
+			staleMeta := &fwkdl.EndpointMetadata{NamespacedName: id, Address: "10.0.0.1"}
+			ds.pods.Store(id, fwkdl.NewEndpoint(staleMeta, fwkdl.NewMetrics()))
+			return nil
+		},
+	}
+	ds = NewDatastore(ctx, factory).(*datastore)
+
+	created, err := ds.upsertEndpoint(ctx, &fwkdl.EndpointMetadata{NamespacedName: id, Address: "10.0.0.2"})
+
+	require.NoError(t, err)
+	assert.False(t, created)
+	eps := ds.PodList(AllPodsPredicate)
+	require.Len(t, eps, 1)
+	assert.Equal(t, "10.0.0.2", eps[0].GetMetadata().Address)
+}
+
+// TestDatastore_ConcurrentAddRemoveCompleteness hammers concurrent delete/upsert of the same pod
+// and asserts the completeness invariant: an upsert either stores the pod or returns an error
+// the caller can retry on (#2060). The fake mirrors Runtime's collector registry: NewEndpoint
+// fails while a collector is registered for the endpoint, and ReleaseEndpoint deregisters only
+// after a delay, so an upsert can observe a pods-map miss while the collector is still
+// registered.
+func TestDatastore_ConcurrentAddRemoveCompleteness(t *testing.T) {
+	ctx := context.Background()
+
+	var mu sync.Mutex
+	registered := map[types.NamespacedName]bool{}
+	factory := &datalayer.FakeEndpointFactory{
+		NewEndpointFn: func(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint {
+			mu.Lock()
+			defer mu.Unlock()
+			if registered[meta.NamespacedName] {
+				return nil
+			}
+			registered[meta.NamespacedName] = true
+			return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics())
+		},
+		ReleaseEndpointFn: func(ep fwkdl.Endpoint) {
+			time.Sleep(50 * time.Microsecond)
+			mu.Lock()
+			defer mu.Unlock()
+			delete(registered, ep.GetMetadata().NamespacedName)
+		},
+	}
+	ds := NewDatastore(ctx, factory)
+	require.NoError(t, ds.PoolSet(ctx, fake.NewFakeClient(), poolutil.InferencePoolToEndpointPool(inferencePool)))
+
+	for i := 0; i < 200; i++ {
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			ds.PodDelete(pod1.Name)
+		}()
+		go func() {
+			defer wg.Done()
+			// The racing upsert may observe the dropped-registration error; the retry below is
+			// what must converge.
+			_ = ds.PodUpdateOrAddIfNotExist(ctx, pod1)
+		}()
+		wg.Wait()
+
+		// The reconciler's requeue loop: retry until the registration lands.
+		require.Eventually(t, func() bool {
+			return ds.PodUpdateOrAddIfNotExist(ctx, pod1) == nil
+		}, 5*time.Second, time.Millisecond, "iteration %d: upsert never converged", i)
+		require.Len(t, ds.PodList(AllPodsPredicate), 1, "iteration %d: pod missing after successful upsert", i)
+	}
+}
+
 func TestDiscoveryNotifier_WorksAlongsideDirectUpsert(t *testing.T) {
 	ctx := context.Background()
 	ds := NewDatastore(ctx, &mockEndpointFactory{})
diff --git a/pkg/epp/requestcontrol/director_test.go b/pkg/epp/requestcontrol/director_test.go
index 7c79bdcdba..d1b2677510 100644
--- a/pkg/epp/requestcontrol/director_test.go
+++ b/pkg/epp/requestcontrol/director_test.go
@@ -888,7 +888,7 @@ func TestDirector_HandleRequest(t *testing.T) {
 					Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 				},
 			}
-			ds.PodUpdateOrAddIfNotExist(ctx, testPod)
+			_ = ds.PodUpdateOrAddIfNotExist(ctx, testPod)
 		}
 
 		for _, test := range tests {
@@ -1055,7 +1055,7 @@ func TestGetRandomEndpoint(t *testing.T) {
 					t.Errorf("unexpected error setting pool: %s", err)
 				}
 				for _, pod := range test.storePods {
-					ds.PodUpdateOrAddIfNotExist(context.Background(), pod)
+					_ = ds.PodUpdateOrAddIfNotExist(context.Background(), pod)
 				}
 				d := &Director{datastore: ds}
 				gotEndpoint := d.GetRandomEndpoint()
@@ -1887,7 +1887,7 @@ func newConditionalDecodeDirector(t *testing.T, scheduleResult *fwksched.Schedul
 	if err := ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(pool)); err != nil {
 		t.Fatalf("PoolSet: %v", err)
 	}
-	ds.PodUpdateOrAddIfNotExist(ctx, &corev1.Pod{
+	_ = ds.PodUpdateOrAddIfNotExist(ctx, &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "default", Labels: map[string]string{"app": "inference"}},
 		Status: corev1.PodStatus{
 			PodIP:      "192.168.1.100",
diff --git a/test/utils/server.go b/test/utils/server.go
index c0feccebf5..710277daa0 100644
--- a/test/utils/server.go
+++ b/test/utils/server.go
@@ -59,7 +59,9 @@ func PrepareForTestStreamingServer(t *testing.T, objectives []*v1alpha2.Inferenc
 	}
 	for _, pod := range pods {
 		initObjs = append(initObjs, pod)
-		ds.PodUpdateOrAddIfNotExist(ctx, pod)
+		if err := ds.PodUpdateOrAddIfNotExist(ctx, pod); err != nil {
+			t.Fatalf("failed to add pod %s to datastore: %v", pod.Name, err)
+		}
 	}
 
 	scheme := runtime.NewScheme()