diff --git a/pkg/virtualkubelet/kept_containers.go b/pkg/virtualkubelet/kept_containers.go
index dd0b26d3..b727fdd5 100644
--- a/pkg/virtualkubelet/kept_containers.go
+++ b/pkg/virtualkubelet/kept_containers.go
@@ -159,7 +159,7 @@ func (p *Provider) currentShadowPod(ctx context.Context, identity shadowResource
 	var newest *v1.Pod
 	for i := range pods.Items {
 		pod := &pods.Items[i]
-		if pod.DeletionTimestamp != nil {
+		if pod.DeletionTimestamp != nil || !isShadowPodOf(pod, identity.OwnerUID) {
 			continue
 		}
 		if newest == nil || newest.CreationTimestamp.Before(&pod.CreationTimestamp) {
diff --git a/pkg/virtualkubelet/kept_containers_test.go b/pkg/virtualkubelet/kept_containers_test.go
index c600a678..f43da6d6 100644
--- a/pkg/virtualkubelet/kept_containers_test.go
+++ b/pkg/virtualkubelet/kept_containers_test.go
@@ -270,6 +270,28 @@ func TestWaitForKeptInitContainers(t *testing.T) {
 		assert.Equal(t, "/delete", path.Load())
 	})
 
+	// The recreated pod took the shadow over in place: the previous pod's shadow pod,
+	// whose kept init container already completed, must not count for this pod.
+	t.Run("a shadow pod of the previous pod with the same name does not count", func(t *testing.T) {
+		pod := podKeepingInit(v1.RestartPolicyNever)
+		previous := shadowOf(t, pod, exited(0))
+		previous.Labels[shadowOfLabel] = "uid-previous-generation"
+		client := fake.NewSimpleClientset(previous)
+		p := keptProvider(client, pod)
+
+		ctx := p.trackKeptInitWait(t.Context(), pod)
+		done := make(chan error, 1)
+		go func() { done <- p.waitForKeptInitContainers(ctx, pod) }()
+
+		select {
+		case err := <-done:
+			t.Fatalf("returned on the previous pod's shadow: %v", err)
+		case <-time.After(100 * time.Millisecond):
+		}
+		p.stopKeptInitWait(string(pod.UID))
+		<-done
+	})
+
 	t.Run("no kept init container: nothing to wait for", func(t *testing.T) {
 		pod := podKeepingInit(v1.RestartPolicyNever)
 		delete(pod.Annotations, annDisableOffloadInitContainers)
diff --git a/pkg/virtualkubelet/shadow.go b/pkg/virtualkubelet/shadow.go
index 88470a9c..71b27412 100644
--- a/pkg/virtualkubelet/shadow.go
+++ b/pkg/virtualkubelet/shadow.go
@@ -159,6 +159,21 @@ func uniqueTruncate(s string, maxLen int, full string) string {
 type shadowResourceIdentity struct {
 	Name      string
 	Namespace string
+	// OwnerUID is the UID of the pod the shadow serves. Shadow resources are named
+	// after the pod, so a pod recreated under the same name reuses those names; the
+	// UID, stamped as a label, tells which pod a resource belongs to.
+	OwnerUID string
+	// PodName and PodNamespace name that pod.
+	PodName      string
+	PodNamespace string
+}
+
+// ownerPod is the namespace/name stamped as shadowOwnerPodAnnotation.
+func (identity shadowResourceIdentity) ownerPod() string {
+	if identity.PodName == "" {
+		return ""
+	}
+	return identity.PodNamespace + "/" + identity.PodName
 }
 
 func isShadowSameNamespace(pod *v1.Pod) bool {
@@ -183,7 +198,10 @@ func computeShadowResourceIdentity(pod *v1.Pod) (shadowResourceIdentity, error)
 		name, namespace = computeShadowResourceNames(pod.Name, pod.Namespace)
 	}
 
-	identity := shadowResourceIdentity{Name: name, Namespace: namespace}
+	identity := shadowResourceIdentity{
+		Name: name, Namespace: namespace,
+		OwnerUID: string(pod.UID), PodName: pod.Name, PodNamespace: pod.Namespace,
+	}
 	if len(fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) > 63 {
 		return shadowResourceIdentity{}, fmt.Errorf("shadow ingress hostname label %q exceeds 63 characters; shorten pod/namespace or disable interlink.eu/shadow-same-ns", fmt.Sprintf("%s-%s", identity.Name, identity.Namespace))
 	}
@@ -301,7 +319,9 @@ func shadowNodeConfigMapName(shadowName string) string {
 // resetShadowNodeConfigMap creates, or blanks, the per-shadow node ConfigMap.
 // The shadow is rendered before the remote batch system has allocated anything,
 // so the ConfigMap has to exist - and be mountable - while still empty. Blanking
-// an existing one stops a previous run's node from being tunnelled to.
+// an existing one stops a previous run's node from being tunnelled to. The cleanup
+// of a deleted pod with the same name can remove an existing one in between, in
+// which case it is created again.
 func (p *Provider) resetShadowNodeConfigMap(ctx context.Context, identity shadowResourceIdentity) error {
 	cm := &v1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -310,18 +330,25 @@ func (p *Provider) resetShadowNodeConfigMap(ctx context.Context, identity shadow
 		},
 		Data: map[string]string{shadowNodeNameKey: "", shadowSocketDirKey: ""},
 	}
+	stampShadowOwner(cm, identity.OwnerUID, identity.ownerPod())
 
-	_, err := p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Create(ctx, cm, metav1.CreateOptions{})
-	if err == nil {
-		return nil
-	}
-	if !apierrors.IsAlreadyExists(err) {
-		return fmt.Errorf("failed to create shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err)
-	}
-	if _, err := p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Update(ctx, cm, metav1.UpdateOptions{}); err != nil {
-		return fmt.Errorf("failed to reset shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err)
+	configMaps := p.clientSet.CoreV1().ConfigMaps(identity.Namespace)
+	for attempt := 0; ; attempt++ {
+		_, err := configMaps.Create(ctx, cm, metav1.CreateOptions{})
+		if err == nil {
+			return nil
+		}
+		if !apierrors.IsAlreadyExists(err) {
+			return fmt.Errorf("failed to create shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err)
+		}
+		_, err = configMaps.Update(ctx, cm, metav1.UpdateOptions{})
+		if err == nil {
+			return nil
+		}
+		if !apierrors.IsNotFound(err) || attempt == 2 {
+			return fmt.Errorf("failed to reset shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err)
+		}
 	}
-	return nil
 }
 
 // publishShadowNode records the remote compute node a pod was allocated on, and
@@ -333,7 +360,9 @@ func (p *Provider) resetShadowNodeConfigMap(ctx context.Context, identity shadow
 // patch, so the shadow never sees a new node with the previous job's directory.
 //
 // Repeated calls with unchanged values are dropped, so the status loop does not
-// write on every poll.
+// write on every poll. A ConfigMap stamped with another pod's UID belongs to a pod
+// recreated under the same name and is left alone; the patch carries the
+// resourceVersion read, so it fails rather than overwrite a change made meanwhile.
 func (p *Provider) publishShadowNode(ctx context.Context, pod *v1.Pod, nodeName, socketDir string) {
 	if nodeName == "" || !p.hasShadow(pod) {
 		return
@@ -361,19 +390,34 @@ func (p *Provider) publishShadowNode(ctx context.Context, pod *v1.Pod, nodeName,
 		return
 	}
 
-	patch, err := json.Marshal(map[string]map[string]string{"data": {
-		shadowNodeNameKey:  nodeName,
-		shadowSocketDirKey: socketDir,
-	}})
+	name := shadowNodeConfigMapName(identity.Name)
+	configMaps := p.clientSet.CoreV1().ConfigMaps(identity.Namespace)
+	current, err := configMaps.Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		log.G(ctx).Warningf("Failed to read shadow configmap %s/%s: %v", identity.Namespace, name, err)
+		return
+	}
+	if owner := current.Labels[shadowOwnerUIDLabel]; owner != "" && identity.OwnerUID != "" && owner != identity.OwnerUID {
+		log.G(ctx).Infof("Not publishing compute node %q for %s/%s: shadow configmap %s belongs to pod UID %s",
+			nodeName, pod.Namespace, pod.Name, name, owner)
+		return
+	}
+
+	var nodePatch struct {
+		Metadata struct {
+			ResourceVersion string `json:"resourceVersion"`
+		} `json:"metadata"`
+		Data map[string]string `json:"data"`
+	}
+	nodePatch.Metadata.ResourceVersion = current.ResourceVersion
+	nodePatch.Data = map[string]string{shadowNodeNameKey: nodeName, shadowSocketDirKey: socketDir}
+	patch, err := json.Marshal(nodePatch)
 	if err != nil {
 		log.G(ctx).Warningf("Failed to marshal shadow node patch for %s/%s: %v", pod.Namespace, pod.Name, err)
 		return
 	}
 
-	name := shadowNodeConfigMapName(identity.Name)
-	_, err = p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Patch(
-		ctx, name, k8stypes.StrategicMergePatchType, patch, metav1.PatchOptions{},
-	)
+	_, err = configMaps.Patch(ctx, name, k8stypes.StrategicMergePatchType, patch, metav1.PatchOptions{})
 	if err != nil {
 		log.G(ctx).Warningf("Failed to publish compute node %q to shadow configmap %s/%s: %v", nodeName, identity.Namespace, name, err)
 		return
@@ -404,10 +448,43 @@ func (p *Provider) releaseShadowOfEndedPod(ctx context.Context, pod *v1.Pod) {
 		return
 	}
 	log.G(ctx).Infof("Pod %s/%s is %s: removing its shadow", pod.Namespace, pod.Name, pod.Status.Phase)
-	p.cleanupShadowResources(ctx, identity.Name, identity.Namespace)
+	p.cleanupShadowResources(ctx, identity)
 	p.forgetShadowNodeName(pod)
 }
 
+// isShadowPodOf reports whether a shadow pod may serve the pod with UID podUID: it is
+// labelled for that pod, or it predates shadowOfLabel, or the UID is unknown.
+func isShadowPodOf(shadowPod *v1.Pod, podUID string) bool {
+	of := shadowPod.Labels[shadowOfLabel]
+	return of == "" || podUID == "" || of == podUID
+}
+
+// ownerIsCurrentPod reports whether the pod a shadow resource is stamped for still
+// exists: the pod named by its shadowOwnerPodAnnotation (the pod of identity when it
+// has none) has the UID of its shadowOwnerUIDLabel. The name comes from the resource
+// because two pods can share a shadow name: sanitising maps web.0 and web-0 to one.
+// Otherwise the owner is gone and its cleanup never ran (it was deleted while the
+// virtual kubelet was down, say), so its leftovers are deleted. When the pod cannot
+// be read, the resource is kept.
+func (p *Provider) ownerIsCurrentPod(ctx context.Context, identity shadowResourceIdentity, obj metav1.Object) bool {
+	namespace, name, found := strings.Cut(obj.GetAnnotations()[shadowOwnerPodAnnotation], "/")
+	if !found {
+		namespace, name = identity.PodNamespace, identity.PodName
+	}
+	if name == "" {
+		return true
+	}
+	pod, err := p.clientSet.CoreV1().Pods(namespace).Get(ctx, name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		return false
+	}
+	if err != nil {
+		log.G(ctx).Warningf("Failed to read pod %s/%s: %v", namespace, name, err)
+		return true
+	}
+	return string(pod.UID) == obj.GetLabels()[shadowOwnerUIDLabel]
+}
+
 // indexShadowPods groups the shadow pods by the UID of the pod they stand for, read
 // from shadowOfLabel. The pods of a shadow rendered before that label existed are
 // left out, so their virtual pod keeps the IP it was created with.
diff --git a/pkg/virtualkubelet/shadow_pod_test.go b/pkg/virtualkubelet/shadow_pod_test.go
index 69c71747..5bf89396 100644
--- a/pkg/virtualkubelet/shadow_pod_test.go
+++ b/pkg/virtualkubelet/shadow_pod_test.go
@@ -41,7 +41,7 @@ func TestWaitForDeploymentPodIgnoresPodsOfAPreviousDeployment(t *testing.T) {
 
 	p := &Provider{clientSet: fake.NewSimpleClientset(deployment, previous, current)}
 
-	got, err := p.waitForDeploymentPod(context.Background(), name, testNamespace)
+	got, err := p.waitForDeploymentPod(context.Background(), name, testNamespace, "")
 	require.NoError(t, err)
 	assert.Equal(t, current.Name, got.Name)
 }
@@ -61,7 +61,29 @@ func TestWaitForDeploymentPodIgnoresTerminatingPods(t *testing.T) {
 
 	p := &Provider{clientSet: fake.NewSimpleClientset(deployment, terminating, current)}
 
-	got, err := p.waitForDeploymentPod(context.Background(), name, testNamespace)
+	got, err := p.waitForDeploymentPod(context.Background(), name, testNamespace, "")
+	require.NoError(t, err)
+	assert.Equal(t, current.Name, got.Name)
+}
+
+// When the recreated pod takes the Deployment over in place, the Deployment keeps its
+// creation time, so the previous pod's shadow pod is not older than it. The label
+// naming the pod each shadow pod stands for tells them apart.
+func TestWaitForDeploymentPodIgnoresPodsOfAnotherPod(t *testing.T) {
+	const name, previousUID, currentUID = "shadow-nb-0-ns", "uid-p1", "uid-p2"
+	now := time.Now()
+
+	deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{
+		Name: name, Namespace: testNamespace, CreationTimestamp: metav1.NewTime(now.Add(-time.Hour)),
+	}}
+	previous := shadowTestPod(name+"-6d5c666bc8-zzzzz", name, "10.0.0.146", now)
+	previous.Labels[shadowOfLabel] = previousUID
+	current := shadowTestPod(name+"-7f8d9a0b1c-aaaaa", name, "10.0.0.141", now.Add(-time.Minute))
+	current.Labels[shadowOfLabel] = currentUID
+
+	p := &Provider{clientSet: fake.NewSimpleClientset(deployment, previous, current)}
+
+	got, err := p.waitForDeploymentPod(context.Background(), name, testNamespace, currentUID)
 	require.NoError(t, err)
 	assert.Equal(t, current.Name, got.Name)
 }
diff --git a/pkg/virtualkubelet/shadow_test.go b/pkg/virtualkubelet/shadow_test.go
index 9282c9e0..a112c530 100644
--- a/pkg/virtualkubelet/shadow_test.go
+++ b/pkg/virtualkubelet/shadow_test.go
@@ -1,15 +1,20 @@
 package virtualkubelet
 
 import (
+	"errors"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
@@ -79,6 +84,30 @@ func TestResetShadowNodeConfigMap(t *testing.T) {
 			"a stale node would make the shadow tunnel to a host whose job is gone")
 		assert.Equal(t, "", cm.Data[shadowSocketDirKey])
 	})
+
+	// The cleanup of a deleted pod with the same name removes its ConfigMap between
+	// this Create (AlreadyExists) and the Update.
+	t.Run("creates it again when it is deleted before the update", func(t *testing.T) {
+		const previousOwner, owner = "uid-previous", "uid-current"
+		client := fake.NewSimpleClientset(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{
+			Name: shadowNodeConfigMapName(identity.Name), Namespace: identity.Namespace,
+			Labels: map[string]string{shadowOwnerUIDLabel: previousOwner},
+		}})
+		gvr := v1.SchemeGroupVersion.WithResource("configmaps")
+		client.PrependReactor("update", "configmaps", func(k8stesting.Action) (bool, runtime.Object, error) {
+			require.NoError(t, client.Tracker().Delete(gvr, identity.Namespace, shadowNodeConfigMapName(identity.Name)))
+			return true, nil, apierrors.NewNotFound(gvr.GroupResource(), shadowNodeConfigMapName(identity.Name))
+		})
+		p := tunnelProvider(client)
+		current := identity
+		current.OwnerUID = owner
+
+		require.NoError(t, p.resetShadowNodeConfigMap(t.Context(), current))
+
+		cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), shadowNodeConfigMapName(identity.Name), metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.Equal(t, owner, cm.Labels[shadowOwnerUIDLabel])
+	})
 }
 
 func TestPublishShadowNode(t *testing.T) {
@@ -184,6 +213,35 @@ func TestPublishShadowNode(t *testing.T) {
 		assert.Empty(t, configMapPatches(client))
 	})
 
+	// A deleted pod stays tracked while its remote deletion runs, after its cache entry
+	// is forgotten: it must not write its node into its same-name successor's ConfigMap.
+	t.Run("a configmap of a pod recreated under the same name is left alone", func(t *testing.T) {
+		client := fake.NewSimpleClientset(&v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{Name: cmName, Namespace: identity.Namespace,
+				Labels: map[string]string{shadowOwnerUIDLabel: "uid-successor"}},
+			Data: map[string]string{shadowNodeNameKey: ""},
+		})
+		p := tunnelProvider(client)
+
+		p.publishShadowNode(t.Context(), pod, testComputeNode, testSocketDir)
+
+		assert.Empty(t, configMapPatches(client))
+	})
+
+	t.Run("the patch carries the resource version that was read", func(t *testing.T) {
+		client := fake.NewSimpleClientset(&v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{Name: cmName, Namespace: identity.Namespace, ResourceVersion: "7"},
+			Data:       map[string]string{shadowNodeNameKey: ""},
+		})
+		p := tunnelProvider(client)
+
+		p.publishShadowNode(t.Context(), pod, testComputeNode, "")
+
+		patches := configMapPatches(client)
+		require.Len(t, patches, 1)
+		assert.Contains(t, string(patches[0].(k8stesting.PatchAction).GetPatch()), `"resourceVersion":"7"`)
+	})
+
 	t.Run("forgetting a pod lets the same node be published again", func(t *testing.T) {
 		client := newClient()
 		p := tunnelProvider(client)
@@ -205,7 +263,7 @@ func TestCleanupShadowResourcesRemovesNodeConfigMap(t *testing.T) {
 	})
 	p := tunnelProvider(client)
 
-	p.cleanupShadowResources(t.Context(), name, ns)
+	p.cleanupShadowResources(t.Context(), shadowResourceIdentity{Name: name, Namespace: ns})
 
 	_, err := client.CoreV1().ConfigMaps(ns).Get(t.Context(), shadowNodeConfigMapName(name), metav1.GetOptions{})
 	assert.True(t, apierrors.IsNotFound(err), "the node configmap should be deleted with the rest of the shadow")
@@ -387,3 +445,324 @@ func TestReleaseShadowIgnoresPodsWithoutAShadow(t *testing.T) {
 
 	assert.Zero(t, deletes(client))
 }
+
+// TestCleanupKeepsShadowOfRecreatedPod covers a pod deleted and recreated under the same
+// name: the new pod's shadow has the same names, and the old pod's cleanup can run after
+// it was created. Only resources stamped with the old pod's UID may go.
+func TestCleanupKeepsShadowOfRecreatedPod(t *testing.T) {
+	const name, podName = testPodDefaultName, "nb-0"
+	const oldUID, newUID = "uid-old", "uid-new"
+	ns := testNamespaceDefault
+	// Every pod is named podName: the successor (newUID) exists, the old pod is gone.
+	cleanup := func(uid string) shadowResourceIdentity {
+		return shadowResourceIdentity{Name: name, Namespace: ns, OwnerUID: uid, PodName: podName, PodNamespace: ns}
+	}
+	owned := func(objectName, uid string) metav1.ObjectMeta {
+		meta := metav1.ObjectMeta{Name: objectName, Namespace: ns}
+		stampShadowOwner(&meta, uid, ns+"/"+podName)
+		return meta
+	}
+	withSuccessor := func(objects ...runtime.Object) *fake.Clientset {
+		successor := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName, Namespace: ns, UID: k8stypes.UID(newUID)}}
+		return fake.NewSimpleClientset(append(objects, successor)...)
+	}
+	stamped := func(uid string) *appsv1.Deployment {
+		return &appsv1.Deployment{ObjectMeta: owned(name, uid)}
+	}
+
+	t.Run("the old pod's cleanup keeps the new pod's shadow", func(t *testing.T) {
+		client := withSuccessor(stamped(newUID))
+		p := tunnelProvider(client)
+
+		p.cleanupShadowResources(t.Context(), cleanup(oldUID))
+
+		_, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})
+		assert.NoError(t, err)
+	})
+
+	t.Run("the owner's cleanup removes it", func(t *testing.T) {
+		client := withSuccessor(stamped(newUID))
+		p := tunnelProvider(client)
+
+		p.cleanupShadowResources(t.Context(), cleanup(newUID))
+
+		_, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})
+		assert.True(t, apierrors.IsNotFound(err))
+	})
+
+	t.Run("an unstamped shadow from an older virtual kubelet is removed", func(t *testing.T) {
+		client := withSuccessor(&appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}})
+		p := tunnelProvider(client)
+
+		p.cleanupShadowResources(t.Context(), cleanup(oldUID))
+
+		_, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})
+		assert.True(t, apierrors.IsNotFound(err))
+	})
+
+	// cert-manager creates the TLS Secret without the owner label: it follows its Ingress.
+	tlsObjects := func(owner string) []runtime.Object {
+		return []runtime.Object{
+			&networkingv1.Ingress{ObjectMeta: owned(name, owner)},
+			&v1.Secret{ObjectMeta: metav1.ObjectMeta{Name: name + "-tls", Namespace: ns}},
+		}
+	}
+	for _, tc := range []struct {
+		cleanup string
+		kept    bool
+	}{{cleanup: oldUID, kept: true}, {cleanup: newUID, kept: false}} {
+		t.Run("the TLS secret follows its ingress, cleanup by "+tc.cleanup, func(t *testing.T) {
+			client := withSuccessor(tlsObjects(newUID)...)
+			p := tunnelProvider(client)
+			p.config.Network.IngressTLS = true
+
+			p.cleanupShadowResources(t.Context(), cleanup(tc.cleanup))
+
+			_, err := client.CoreV1().Secrets(ns).Get(t.Context(), name+"-tls", metav1.GetOptions{})
+			if tc.kept {
+				assert.NoError(t, err)
+			} else {
+				assert.True(t, apierrors.IsNotFound(err))
+			}
+		})
+	}
+
+	// conflictOnce makes the first delete of resource fail with a conflict, after
+	// changing the stored object with change, as an update landing between read and delete.
+	conflictOnce := func(client *fake.Clientset, resource string, gvr schema.GroupVersionResource, change func(metav1.Object)) {
+		first := true
+		client.PrependReactor("delete", resource, func(k8stesting.Action) (bool, runtime.Object, error) {
+			if !first {
+				return false, nil, nil
+			}
+			first = false
+			obj, err := client.Tracker().Get(gvr, ns, name)
+			require.NoError(t, err)
+			change(obj.(metav1.Object))
+			require.NoError(t, client.Tracker().Update(gvr, obj, ns))
+			return true, nil, apierrors.NewConflict(gvr.GroupResource(), name, errors.New("resourceVersion changed"))
+		})
+	}
+
+	t.Run("an ingress taken over between read and delete keeps it and its TLS secret", func(t *testing.T) {
+		client := withSuccessor(tlsObjects(oldUID)...)
+		conflictOnce(client, "ingresses", networkingv1.SchemeGroupVersion.WithResource("ingresses"), func(o metav1.Object) {
+			o.SetLabels(map[string]string{shadowOwnerUIDLabel: newUID})
+		})
+		p := tunnelProvider(client)
+		p.config.Network.IngressTLS = true
+
+		p.cleanupShadowResources(t.Context(), cleanup(oldUID))
+
+		_, err := client.NetworkingV1().Ingresses(ns).Get(t.Context(), name, metav1.GetOptions{})
+		assert.NoError(t, err)
+		_, err = client.CoreV1().Secrets(ns).Get(t.Context(), name+"-tls", metav1.GetOptions{})
+		assert.NoError(t, err)
+	})
+
+	t.Run("any other change between read and delete is retried", func(t *testing.T) {
+		client := withSuccessor(stamped(oldUID))
+		conflictOnce(client, "deployments", appsv1.SchemeGroupVersion.WithResource("deployments"), func(o metav1.Object) {
+			o.SetAnnotations(map[string]string{"status": "updated"})
+		})
+		p := tunnelProvider(client)
+
+		p.cleanupShadowResources(t.Context(), cleanup(oldUID))
+
+		_, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})
+		assert.True(t, apierrors.IsNotFound(err))
+	})
+
+	t.Run("the TLS secret stays when its ingress cannot be read", func(t *testing.T) {
+		client := withSuccessor(tlsObjects(newUID)...)
+		client.PrependReactor("get", "ingresses", func(k8stesting.Action) (bool, runtime.Object, error) {
+			return true, nil, apierrors.NewInternalError(errors.New("etcd timeout"))
+		})
+		p := tunnelProvider(client)
+		p.config.Network.IngressTLS = true
+
+		p.cleanupShadowResources(t.Context(), cleanup(oldUID))
+
+		_, err := client.CoreV1().Secrets(ns).Get(t.Context(), name+"-tls", metav1.GetOptions{})
+		assert.NoError(t, err)
+	})
+
+	t.Run("the delete is conditional on the resource version that was read", func(t *testing.T) {
+		deployment := stamped(oldUID)
+		deployment.UID, deployment.ResourceVersion = "deployment-uid", "41"
+		client := withSuccessor(deployment)
+		var preconditions *metav1.Preconditions
+		client.PrependReactor("delete", "deployments", func(action k8stesting.Action) (bool, runtime.Object, error) {
+			preconditions = action.(k8stesting.DeleteAction).GetDeleteOptions().Preconditions
+			return false, nil, nil
+		})
+		p := tunnelProvider(client)
+
+		p.cleanupShadowResources(t.Context(), cleanup(oldUID))
+
+		require.NotNil(t, preconditions)
+		require.NotNil(t, preconditions.UID)
+		require.NotNil(t, preconditions.ResourceVersion)
+		assert.Equal(t, k8stypes.UID("deployment-uid"), *preconditions.UID)
+		assert.Equal(t, "41", *preconditions.ResourceVersion)
+	})
+}
+
+// A resource stamped with another pod's UID is kept only while that pod is the one
+// that has the name; the leftovers of a pod whose cleanup never ran are deleted.
+func TestCleanupKeepsOnlyTheCurrentPodsResources(t *testing.T) {
+	const name, podName, staleUID, currentUID = testPodDefaultName, "nb-0", "uid-stale", "uid-live"
+	ns := testNamespaceDefault
+	identity := shadowResourceIdentity{Name: name, Namespace: ns, OwnerUID: "uid-cleaning", PodName: podName, PodNamespace: ns}
+	stampedService := func(owner, ownerPod string) *v1.Service {
+		meta := metav1.ObjectMeta{Name: name, Namespace: ns}
+		stampShadowOwner(&meta, owner, ownerPod)
+		return &v1.Service{ObjectMeta: meta}
+	}
+	podIn := func(namespace, podName, uid string) *v1.Pod {
+		return &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName, Namespace: namespace, UID: k8stypes.UID(uid)}}
+	}
+	pod := func(podName, uid string) *v1.Pod { return podIn(ns, podName, uid) }
+	serviceLeft := func(t *testing.T, client *fake.Clientset) bool {
+		t.Helper()
+		tunnelProvider(client).cleanupShadowResources(t.Context(), identity)
+		_, err := client.CoreV1().Services(ns).Get(t.Context(), name, metav1.GetOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			t.Fatal(err)
+		}
+		return err == nil
+	}
+
+	t.Run("kept for the pod that now has the name", func(t *testing.T) {
+		client := fake.NewSimpleClientset(stampedService(currentUID, ns+"/"+podName), pod(podName, currentUID))
+		assert.True(t, serviceLeft(t, client))
+	})
+
+	t.Run("deleted when another pod now has the name", func(t *testing.T) {
+		client := fake.NewSimpleClientset(stampedService(staleUID, ns+"/"+podName), pod(podName, currentUID))
+		assert.False(t, serviceLeft(t, client), "a leftover of a pod that is gone stays forever otherwise")
+	})
+
+	t.Run("deleted when no pod has the name", func(t *testing.T) {
+		client := fake.NewSimpleClientset(stampedService(staleUID, ns+"/"+podName))
+		assert.False(t, serviceLeft(t, client))
+	})
+
+	t.Run("kept when the pod cannot be read", func(t *testing.T) {
+		client := fake.NewSimpleClientset(stampedService(staleUID, ns+"/"+podName))
+		client.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) {
+			return true, nil, apierrors.NewInternalError(errors.New("etcd timeout"))
+		})
+		assert.True(t, serviceLeft(t, client))
+	})
+
+	// Stamped before the annotation existed: the owner is looked up under the
+	// cleaning pod's name.
+	t.Run("without the annotation, deleted when another pod now has the name", func(t *testing.T) {
+		client := fake.NewSimpleClientset(stampedService(staleUID, ""), pod(podName, currentUID))
+		assert.False(t, serviceLeft(t, client))
+	})
+
+	// Namespaces a--b and a-b get the same shadow names too.
+	t.Run("kept for a live pod of another namespace that shares the shadow name", func(t *testing.T) {
+		const otherNamespace, otherUID = "other-ns", "uid-elsewhere"
+		client := fake.NewSimpleClientset(stampedService(otherUID, otherNamespace+"/"+podName),
+			pod(podName, currentUID), podIn(otherNamespace, podName, otherUID))
+		assert.True(t, serviceLeft(t, client))
+	})
+
+	// web.0 and web-0 get the same shadow name: deleting one must not take the
+	// shadow the other one, still running, took over.
+	t.Run("kept for a live pod of another name that shares the shadow name", func(t *testing.T) {
+		const other, otherUID = "nb.0", "uid-other"
+		client := fake.NewSimpleClientset(stampedService(otherUID, ns+"/"+other), pod(podName, currentUID), pod(other, otherUID))
+		assert.True(t, serviceLeft(t, client))
+	})
+}
+
+// The cleanup of a deleted pod and the shadow of its same-name successor must not
+// interleave: the new Ingress could otherwise appear between the old Ingress and the
+// TLS Secret being deleted, and lose the Secret.
+func TestShadowCleanupAndApplyAreSerialised(t *testing.T) {
+	const name = testPodDefaultName
+	ns := testNamespaceDefault
+	waitsForLock := func(t *testing.T, p *Provider, run func()) {
+		t.Helper()
+		unlock := p.lockShadow(ns, name)
+		done := make(chan struct{})
+		go func() { run(); close(done) }()
+		select {
+		case <-done:
+			unlock()
+			t.Fatal("ran while the shadow's lock was held")
+		case <-time.After(100 * time.Millisecond):
+		}
+		unlock()
+		select {
+		case <-done:
+		case <-time.After(2 * time.Second):
+			t.Fatal("did not run once the lock was released")
+		}
+	}
+
+	t.Run("cleanup", func(t *testing.T) {
+		client := fake.NewSimpleClientset()
+		p := tunnelProvider(client)
+		waitsForLock(t, p, func() {
+			p.cleanupShadowResources(t.Context(), shadowResourceIdentity{Name: name, Namespace: ns})
+		})
+	})
+
+	t.Run("apply", func(t *testing.T) {
+		client := fake.NewSimpleClientset()
+		p := tunnelProvider(client)
+		manifest := "apiVersion: v1\nkind: ConfigMap\nmetadata:\n  name: " + name + "\n  namespace: " + ns + "\n"
+		waitsForLock(t, p, func() {
+			_, _, err := p.applyShadowObjects(t.Context(), manifest, &ShadowTemplateData{Name: name, Namespace: ns})
+			assert.NoError(t, err)
+		})
+		_, err := client.CoreV1().ConfigMaps(ns).Get(t.Context(), name, metav1.GetOptions{})
+		assert.NoError(t, err)
+	})
+}
+
+// Applying a shadow stamps every object with the pod's UID, and an object left by a
+// deleted pod with the same name is updated in place with the new pod's UID.
+func TestApplyShadowStampsOwner(t *testing.T) {
+	const name, previousOwner, owner = testPodDefaultName, "uid-before", "uid-after"
+	ns := testNamespaceDefault
+	client := fake.NewSimpleClientset(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{
+		Name: name, Namespace: ns, Labels: map[string]string{shadowOwnerUIDLabel: previousOwner},
+	}})
+	p := tunnelProvider(client)
+	manifest := "apiVersion: v1\nkind: ConfigMap\nmetadata:\n  name: " + name + "\n  namespace: " + ns + "\n" +
+		"---\napiVersion: v1\nkind: Service\nmetadata:\n  name: " + name + "\n  namespace: " + ns + "\n" +
+		"spec:\n  ports:\n  - port: 80\n"
+
+	_, _, err := p.applyShadowObjects(t.Context(), manifest, &ShadowTemplateData{Name: name, Namespace: ns, OwnerUID: owner, OwnerPod: ns + "/nb-0"})
+	require.NoError(t, err)
+
+	cm, err := client.CoreV1().ConfigMaps(ns).Get(t.Context(), name, metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, owner, cm.Labels[shadowOwnerUIDLabel], "the previous pod's object is relabelled")
+	assert.Equal(t, ns+"/nb-0", cm.Annotations[shadowOwnerPodAnnotation])
+	svc, err := client.CoreV1().Services(ns).Get(t.Context(), name, metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, owner, svc.Labels[shadowOwnerUIDLabel])
+	assert.Equal(t, ns+"/nb-0", svc.Annotations[shadowOwnerPodAnnotation])
+}
+
+func TestShadowIdentityAndNodeConfigMapCarryOwner(t *testing.T) {
+	pod := podWithPort("nb", testNamespaceDefault, "uid-7")
+	identity, err := computeShadowResourceIdentity(pod)
+	require.NoError(t, err)
+	assert.Equal(t, "uid-7", identity.OwnerUID)
+
+	client := fake.NewSimpleClientset()
+	p := tunnelProvider(client)
+	require.NoError(t, p.resetShadowNodeConfigMap(t.Context(), identity))
+	cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), shadowNodeConfigMapName(identity.Name), metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, "uid-7", cm.Labels[shadowOwnerUIDLabel])
+	assert.Equal(t, testNamespaceDefault+"/nb", cm.Annotations[shadowOwnerPodAnnotation])
+}
diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go
index a1f07406..d0327c47 100644
--- a/pkg/virtualkubelet/virtualkubelet.go
+++ b/pkg/virtualkubelet/virtualkubelet.go
@@ -9,6 +9,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"io"
 	mathrand "math/rand"
 	"net"
@@ -39,6 +40,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/retry"
 	stats "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
 
 	types "github.com/interlink-hq/interlink/pkg/interlink"
@@ -102,6 +104,8 @@ const (
 	annDisableOffloadInitContainers = "interlink.eu/disable-offload-init-containers" // comma-separated init container names
 	annMeshNetworkDisabled          = "interlink.eu/mesh-network"                    // set to "disabled" to opt out of mesh networking
 	annShadowSocketPorts            = "interlink.eu/shadow-socket-ports"             // comma-separated exposed TCP ports served on unix sockets
+	shadowOwnerUIDLabel             = "interlink.eu/shadow-owner-uid"
+	shadowOwnerPodAnnotation        = "interlink.eu/shadow-owner-pod" // namespace/name of the pod in shadowOwnerUIDLabel
 )
 
 type ShadowTemplateData struct {
@@ -130,6 +134,10 @@ type ShadowTemplateData struct {
 	// in a container named PauseName, a name no kept init container uses.
 	PauseImage string
 	PauseName  string
+	// OwnerUID and OwnerPod are stamped on every shadow resource (see
+	// shadowResourceIdentity).
+	OwnerUID string
+	OwnerPod string
 	// NodeConfigMap is the ConfigMap the shadow can mount to learn which remote
 	// compute node the workload landed on. It exists from the moment the shadow is
 	// created but holds an empty NodeNameKey until the plugin reports the node.
@@ -204,7 +212,10 @@ type Provider struct {
 	shadowsReleased sync.Map
 	// keptInitWaits holds, per pod UID, the cancel function of a pending wait for the
 	// pod's kept init containers (see stopKeptInitWait).
-	keptInitWaits    sync.Map
+	keptInitWaits sync.Map
+	// shadowLocks serialise applying a shadow's manifests and cleaning them up, per
+	// shadow name (see lockShadow).
+	shadowLocks      [64]sync.Mutex
 	pendingDeletes   map[string]*pendingDelete
 	pendingDeletesMu sync.Mutex
 	deleteLoopOnce   sync.Once
@@ -1074,6 +1085,8 @@ func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v
 		KeptOnly:             keptOnly,
 		PauseImage:           p.shadowPauseImage(),
 		PauseName:            shadowPauseName(localInitContainers),
+		OwnerUID:             identity.OwnerUID,
+		OwnerPod:             identity.ownerPod(),
 		NodeConfigMap:        shadowNodeConfigMapName(identity.Name),
 		NodeNameKey:          shadowNodeNameKey,
 		SocketDirKey:         shadowSocketDirKey,
@@ -1414,6 +1427,28 @@ func (p *Provider) applyOrUpdateSecret(ctx context.Context, o *v1.Secret) error
 
 // applyShadowManifests applies the generated manifests and returns the first created pod
 func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string, td *ShadowTemplateData) (*v1.Pod, error) {
+	deploymentName, namespace, err := p.applyShadowObjects(ctx, manifestYAML, td)
+	if err != nil {
+		return nil, err
+	}
+	if deploymentName != "" {
+		podUID := ""
+		if td != nil {
+			podUID = td.PodUID
+		}
+		return p.waitForDeploymentPod(ctx, deploymentName, namespace, podUID)
+	}
+
+	return nil, fmt.Errorf("no deployment found in manifests")
+}
+
+// applyShadowObjects creates or updates the objects of the shadow manifests and returns
+// the name and namespace of its Deployment. It holds the shadow's lock meanwhile, so the
+// cleanup of a deleted pod with the same name never runs in between.
+func (p *Provider) applyShadowObjects(ctx context.Context, manifestYAML string, td *ShadowTemplateData) (string, string, error) {
+	if td != nil {
+		defer p.lockShadow(td.Namespace, td.Name)()
+	}
 	resources := strings.Split(manifestYAML, "---")
 	decoder := serializer.NewCodecFactory(scheme.Scheme).UniversalDeserializer()
 	var deploymentName string
@@ -1431,13 +1466,16 @@ func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string
 			log.G(ctx).Warningf("Failed to decode resource: %v", err)
 			continue
 		}
+		if meta, ok := obj.(metav1.Object); ok && td != nil {
+			stampShadowOwner(meta, td.OwnerUID, td.OwnerPod)
+		}
 
 		switch o := obj.(type) {
 		case *appsv1.Deployment:
 			name, ns, err := p.applyOrUpdateDeployment(ctx, o, td)
 			if err != nil {
 				p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace)
-				return nil, err
+				return "", "", err
 			}
 			deploymentName = name
 			namespace = ns
@@ -1446,28 +1484,28 @@ func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string
 		case *v1.Service:
 			if err := p.applyOrUpdateService(ctx, o); err != nil {
 				p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace)
-				return nil, err
+				return "", "", err
 			}
 			createdResources = append(createdResources, "service:"+o.Name)
 
 		case *networkingv1.Ingress:
 			if err := p.applyOrUpdateIngress(ctx, o); err != nil {
 				p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace)
-				return nil, err
+				return "", "", err
 			}
 			createdResources = append(createdResources, "ingress:"+o.Name)
 
 		case *v1.ConfigMap:
 			if err := p.applyOrUpdateConfigMap(ctx, o); err != nil {
 				p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace)
-				return nil, err
+				return "", "", err
 			}
 			createdResources = append(createdResources, "configmap:"+o.Name)
 
 		case *v1.Secret:
 			if err := p.applyOrUpdateSecret(ctx, o); err != nil {
 				p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace)
-				return nil, err
+				return "", "", err
 			}
 			createdResources = append(createdResources, "secret:"+o.Name)
 
@@ -1476,11 +1514,7 @@ func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string
 		}
 	}
 
-	if deploymentName != "" {
-		return p.waitForDeploymentPod(ctx, deploymentName, namespace)
-	}
-
-	return nil, fmt.Errorf("no deployment found in manifests")
+	return deploymentName, namespace, nil
 }
 
 // waitForDeploymentPod waits for a deployment to create a pod and returns it.
@@ -1488,8 +1522,10 @@ func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string
 // A pod recreated under the same name (a StatefulSet replica) gets a Deployment with the
 // same name and pod selector as the one just deleted, and the garbage collector
 // removes that one's pods asynchronously. Pods that are terminating, or older
-// than the current Deployment, belong to the previous one and are skipped.
-func (p *Provider) waitForDeploymentPod(ctx context.Context, deploymentName, namespace string) (*v1.Pod, error) {
+// than the current Deployment, belong to the previous one and are skipped. So are
+// pods labelled for another pod than podUID: when the recreated pod takes the
+// Deployment over in place, the previous pod's shadow pod is newer than it.
+func (p *Provider) waitForDeploymentPod(ctx context.Context, deploymentName, namespace, podUID string) (*v1.Pod, error) {
 	timeout := 30 * time.Second
 	start := time.Now()
 
@@ -1515,6 +1551,9 @@ func (p *Provider) waitForDeploymentPod(ctx context.Context, deploymentName, nam
 			if pod.DeletionTimestamp != nil || pod.CreationTimestamp.Before(&deployment.CreationTimestamp) {
 				continue
 			}
+			if !isShadowPodOf(pod, podUID) {
+				continue
+			}
 			return pod, nil
 		}
 
@@ -1524,61 +1563,119 @@ func (p *Provider) waitForDeploymentPod(ctx context.Context, deploymentName, nam
 	return nil, fmt.Errorf("no pod found for deployment %s within timeout", deploymentName)
 }
 
-// cleanupShadowResources removes all shadow resources for a given name and namespace
-func (p *Provider) cleanupShadowResources(ctx context.Context, shadowName, namespace string) {
+// cleanupShadowResources removes the shadow resources of the pod identity names. A
+// resource stamped with another pod's UID is left alone: it belongs to a pod recreated
+// under the same name, whose shadow can exist before this pod's cleanup runs. The
+// shadow's lock keeps that shadow from being applied while the cleanup runs.
+func (p *Provider) cleanupShadowResources(ctx context.Context, identity shadowResourceIdentity) {
+	shadowName, namespace := identity.Name, identity.Namespace
+	defer p.lockShadow(namespace, shadowName)()
 	log.G(ctx).Infof("Cleaning up shadow resources for %s/%s", namespace, shadowName)
 
-	// Delete deployment
-	err := p.clientSet.AppsV1().Deployments(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{})
-	if err != nil {
-		log.G(ctx).Warningf("Failed to delete shadow deployment %s/%s: %v", namespace, shadowName, err)
-	} else {
-		log.G(ctx).Infof("Successfully deleted shadow deployment %s/%s", namespace, shadowName)
-	}
-
-	// Delete service
-	err = p.clientSet.CoreV1().Services(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{})
-	if err != nil {
-		log.G(ctx).Warningf("Failed to delete shadow service %s/%s: %v", namespace, shadowName, err)
-	} else {
-		log.G(ctx).Infof("Successfully deleted shadow service %s/%s", namespace, shadowName)
-	}
-
-	// Delete ingress
-	err = p.clientSet.NetworkingV1().Ingresses(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{})
-	if err != nil {
-		log.G(ctx).Warningf("Failed to delete shadow ingress %s/%s: %v", namespace, shadowName, err)
-	} else {
-		log.G(ctx).Infof("Successfully deleted shadow ingress %s/%s", namespace, shadowName)
+	deployments := p.clientSet.AppsV1().Deployments(namespace)
+	p.deleteOwnedShadowResource(ctx, identity, "deployment", shadowName,
+		func(name string) (metav1.Object, error) { return deployments.Get(ctx, name, metav1.GetOptions{}) },
+		func(name string, opts metav1.DeleteOptions) error { return deployments.Delete(ctx, name, opts) })
+
+	services := p.clientSet.CoreV1().Services(namespace)
+	p.deleteOwnedShadowResource(ctx, identity, "service", shadowName,
+		func(name string) (metav1.Object, error) { return services.Get(ctx, name, metav1.GetOptions{}) },
+		func(name string, opts metav1.DeleteOptions) error { return services.Delete(ctx, name, opts) })
+
+	ingresses := p.clientSet.NetworkingV1().Ingresses(namespace)
+	ingressKept := p.deleteOwnedShadowResource(ctx, identity, "ingress", shadowName,
+		func(name string) (metav1.Object, error) { return ingresses.Get(ctx, name, metav1.GetOptions{}) },
+		func(name string, opts metav1.DeleteOptions) error { return ingresses.Delete(ctx, name, opts) })
+
+	configMaps := p.clientSet.CoreV1().ConfigMaps(namespace)
+	for _, name := range []string{shadowName + "-wg-config", shadowNodeConfigMapName(shadowName)} {
+		p.deleteOwnedShadowResource(ctx, identity, "configmap", name,
+			func(name string) (metav1.Object, error) { return configMaps.Get(ctx, name, metav1.GetOptions{}) },
+			func(name string, opts metav1.DeleteOptions) error { return configMaps.Delete(ctx, name, opts) })
+	}
+
+	// Delete cert-manager-provisioned TLS secret. cert-manager creates it without the
+	// owner label, so it goes with its Ingress: kept when another pod's Ingress is.
+	if p.config.Network.IngressTLS && !ingressKept {
+		secrets := p.clientSet.CoreV1().Secrets(namespace)
+		p.deleteOwnedShadowResource(ctx, identity, "secret", shadowName+"-tls",
+			func(name string) (metav1.Object, error) { return secrets.Get(ctx, name, metav1.GetOptions{}) },
+			func(name string, opts metav1.DeleteOptions) error { return secrets.Delete(ctx, name, opts) })
+	}
+}
+
+// deleteOwnedShadowResource deletes one shadow resource unless it is stamped with a
+// pod UID other than identity.OwnerUID, and reports whether it may still serve another
+// pod: kept for its owner, changed meanwhile, or left in place by an error. The delete
+// is conditional on the UID and resourceVersion that were read, so a replacement or an
+// in-place update for a recreated pod made in between is never removed: on a conflict
+// the resource is read again and the owner checked again. Unstamped resources (from
+// before the stamp existed) are deleted as before.
+func (p *Provider) deleteOwnedShadowResource(ctx context.Context, identity shadowResourceIdentity, kind, name string,
+	get func(string) (metav1.Object, error), del func(string, metav1.DeleteOptions) error) bool {
+	kept := false
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		obj, err := get(name)
+		if err != nil {
+			return err
+		}
+		if owner := obj.GetLabels()[shadowOwnerUIDLabel]; owner != "" && identity.OwnerUID != "" && owner != identity.OwnerUID &&
+			p.ownerIsCurrentPod(ctx, identity, obj) {
+			log.G(ctx).Infof("Keeping shadow %s %s/%s: it belongs to pod UID %s, not %s", kind, identity.Namespace, name, owner, identity.OwnerUID)
+			kept = true
+			return nil
+		}
+		uid, resourceVersion := obj.GetUID(), obj.GetResourceVersion()
+		return del(name, metav1.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &uid, ResourceVersion: &resourceVersion}})
+	})
+	switch {
+	case err == nil:
+		if !kept {
+			log.G(ctx).Infof("Successfully deleted shadow %s %s/%s", kind, identity.Namespace, name)
+		}
+	case apierrors.IsConflict(err):
+		log.G(ctx).Warningf("Keeping shadow %s %s/%s: it kept changing while being deleted", kind, identity.Namespace, name)
+		kept = true
+	case !apierrors.IsNotFound(err):
+		// Its owner is unknown, so it may serve another pod.
+		log.G(ctx).Warningf("Failed to delete shadow %s %s/%s: %v", kind, identity.Namespace, name, err)
+		kept = true
+	}
+	return kept
+}
+
+// lockShadow locks the shadow named name in namespace and returns the unlock function.
+// The lock is one of a fixed set, chosen by name, so it is shared with unrelated shadows
+// now and then; it only orders a cleanup and an apply within this virtual kubelet.
+func (p *Provider) lockShadow(namespace, name string) func() {
+	h := fnv.New32a()
+	_, _ = h.Write([]byte(namespace + "/" + name))
+	mu := &p.shadowLocks[h.Sum32()%uint32(len(p.shadowLocks))]
+	mu.Lock()
+	return mu.Unlock
+}
+
+// stampShadowOwner labels a shadow resource with the UID of the pod it serves and
+// annotates it with that pod's namespace/name.
+func stampShadowOwner(obj metav1.Object, ownerUID, ownerPod string) {
+	if ownerUID == "" {
+		return
 	}
-
-	// Delete configmap
-	err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, shadowName+"-wg-config", metav1.DeleteOptions{})
-	if err != nil {
-		log.G(ctx).Warningf("Failed to delete shadow configmap %s/%s: %v", namespace, shadowName+"-wg-config", err)
-	} else {
-		log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, shadowName+"-wg-config")
+	labels := obj.GetLabels()
+	if labels == nil {
+		labels = map[string]string{}
 	}
-
-	// Delete the compute node configmap
-	nodeCM := shadowNodeConfigMapName(shadowName)
-	err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, nodeCM, metav1.DeleteOptions{})
-	if err != nil {
-		log.G(ctx).Warningf("Failed to delete shadow configmap %s/%s: %v", namespace, nodeCM, err)
-	} else {
-		log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, nodeCM)
+	labels[shadowOwnerUIDLabel] = ownerUID
+	obj.SetLabels(labels)
+	if ownerPod == "" {
+		return
 	}
-
-	// Delete cert-manager-provisioned TLS secret.
-	if p.config.Network.IngressTLS {
-		secretName := shadowName + "-tls"
-		err = p.clientSet.CoreV1().Secrets(namespace).Delete(ctx, secretName, metav1.DeleteOptions{})
-		if err != nil {
-			log.G(ctx).Warningf("Failed to delete shadow TLS secret %s/%s: %v", namespace, secretName, err)
-		} else {
-			log.G(ctx).Infof("Successfully deleted shadow TLS secret %s/%s", namespace, secretName)
-		}
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
 	}
+	annotations[shadowOwnerPodAnnotation] = ownerPod
+	obj.SetAnnotations(annotations)
 }
 
 // cleanupPartialShadowResources removes specific resources that were created before a failure
@@ -1873,7 +1970,7 @@ func (p *Provider) handleShadowCreation(ctx context.Context, pod *v1.Pod) (strin
 	if err != nil {
 		log.G(ctx).Errorf("Failed to create shadow infrastructure for %s/%s: %v", pod.Namespace, pod.Name, err)
 		// Clean up any partially created resources
-		p.cleanupShadowResources(ctx, identity.Name, identity.Namespace)
+		p.cleanupShadowResources(ctx, identity)
 		return "", fmt.Errorf("failed to create shadow infrastructure for exposed ports: %w", err)
 	}
 
@@ -1887,10 +1984,7 @@ func (p *Provider) handleShadowCreation(ctx context.Context, pod *v1.Pod) (strin
 
 	// This is the address the virtual pod starts with: if the shadow pod is replaced
 	// later, the status loop moves it to the replacement (see refreshShadowPodIP).
-	podIP, err := p.waitForShadowPodIP(ctx, shadowPod, timeout, shadowResourceIdentity{
-		Name:      templateData.Name,
-		Namespace: templateData.Namespace,
-	})
+	podIP, err := p.waitForShadowPodIP(ctx, shadowPod, timeout, identity)
 	if err != nil {
 		log.G(ctx).Errorf("Failed to get shadow pod IP for %s/%s: %v", pod.Namespace, pod.Name, err)
 		return "", err
@@ -1934,7 +2028,7 @@ func (p *Provider) waitForShadowPodIP(ctx context.Context, shadowPod *v1.Pod, ti
 	}
 
 	// Clean up the shadow infrastructure since it didn't get an IP
-	p.cleanupShadowResources(ctx, identity.Name, identity.Namespace)
+	p.cleanupShadowResources(ctx, identity)
 	return "", fmt.Errorf("shadow pod %s/%s failed to get an IP within %v timeout", shadowPod.Namespace, shadowPod.Name, timeout)
 }
 
@@ -2201,7 +2295,7 @@ func (p *Provider) DeletePod(ctx context.Context, pod *v1.Pod) (err error) {
 			if identityErr != nil {
 				log.G(ctx).Warningf("Failed to compute shadow resource identity for %s/%s: %v", pod.Namespace, pod.Name, identityErr)
 			} else {
-				p.cleanupShadowResources(ctx, identity.Name, identity.Namespace)
+				p.cleanupShadowResources(ctx, identity)
 			}
 		}
 		p.forgetShadowNodeName(pod)
diff --git a/pkg/virtualkubelet/virtualkubelet_mesh_test.go b/pkg/virtualkubelet/virtualkubelet_mesh_test.go
index 488d9a8d..d37f5a41 100644
--- a/pkg/virtualkubelet/virtualkubelet_mesh_test.go
+++ b/pkg/virtualkubelet/virtualkubelet_mesh_test.go
@@ -312,7 +312,7 @@ func TestCleanupShadowResources(t *testing.T) {
 		client := newClient()
 		p := &Provider{clientSet: client, config: Config{Network: Network{IngressTLS: true}}}
 
-		p.cleanupShadowResources(t.Context(), name, ns)
+		p.cleanupShadowResources(t.Context(), shadowResourceIdentity{Name: name, Namespace: ns})
 
 		_, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})
 		assert.True(t, apierrors.IsNotFound(err), "deployment should be deleted")
@@ -330,7 +330,7 @@ func TestCleanupShadowResources(t *testing.T) {
 		client := newClient()
 		p := &Provider{clientSet: client, config: Config{Network: Network{IngressTLS: false}}}
 
-		p.cleanupShadowResources(t.Context(), name, ns)
+		p.cleanupShadowResources(t.Context(), shadowResourceIdentity{Name: name, Namespace: ns})
 
 		// Core resources are still removed regardless of TLS...
 		_, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})