diff --git a/lib/auth/auth.go b/lib/auth/auth.go
index 40c831f1dc1f4..c793ff7dc0478 100644
--- a/lib/auth/auth.go
+++ b/lib/auth/auth.go
@@ -542,6 +542,10 @@ type Server struct {
 
 	// license is the Teleport Enterprise license used to start the auth server
 	license *liblicense.License
+
+	// headlessAuthenticationWatcher is a headless authentication watcher,
+	// used to catch and propagate headless authentication request changes.
+	headlessAuthenticationWatcher *local.HeadlessAuthenticationWatcher
 }
 
 // SetSAMLService registers svc as the SAMLService that provides the SAML
@@ -604,6 +608,12 @@ func (a *Server) checkLockInForce(mode constants.LockingMode, targets []types.Lo
 	return a.lockWatcher.CheckLockInForce(mode, targets...)
 }
 
+func (a *Server) SetHeadlessAuthenticationWatcher(headlessAuthenticationWatcher *local.HeadlessAuthenticationWatcher) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.headlessAuthenticationWatcher = headlessAuthenticationWatcher
+}
+
 // runPeriodicOperations runs some periodic bookkeeping operations
 // performed by auth server
 func (a *Server) runPeriodicOperations() {
diff --git a/lib/auth/helpers.go b/lib/auth/helpers.go
index 73537d12ac3d8..613e77a1bc86e 100644
--- a/lib/auth/helpers.go
+++ b/lib/auth/helpers.go
@@ -354,6 +354,14 @@ func NewTestAuthServer(cfg TestAuthServerConfig) (*TestAuthServer, error) {
 	}
 	srv.AuthServer.SetLockWatcher(srv.LockWatcher)
 
+	headlessAuthenticationWatcher, err := local.NewHeadlessAuthenticationWatcher(ctx, local.HeadlessAuthenticationWatcherConfig{
+		Backend: b,
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	srv.AuthServer.SetHeadlessAuthenticationWatcher(headlessAuthenticationWatcher)
+
 	srv.Authorizer, err = authz.NewAuthorizer(authz.AuthorizerOpts{
 		ClusterName: srv.ClusterName,
 		AccessPoint: srv.AuthServer,
diff --git a/lib/service/service.go b/lib/service/service.go
index 07c2371e2a79d..5adb189288cdc 100644
--- a/lib/service/service.go
+++ b/lib/service/service.go
@@ -1589,6 +1589,14 @@ func (process *TeleportProcess) initAuthService() error {
 	}
 	authServer.SetLockWatcher(lockWatcher)
 
+	headlessAuthenticationWatcher, err := local.NewHeadlessAuthenticationWatcher(process.ExitContext(), local.HeadlessAuthenticationWatcherConfig{
+		Backend: b,
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	authServer.SetHeadlessAuthenticationWatcher(headlessAuthenticationWatcher)
+
 	process.setLocalAuth(authServer)
 
 	// The auth server runs its own upload completer, which is necessary in sync recording modes where
diff --git a/lib/services/local/headlessauthn.go b/lib/services/local/headlessauthn.go
index a0f0515e1ed90..a0ad206835a34 100644
--- a/lib/services/local/headlessauthn.go
+++ b/lib/services/local/headlessauthn.go
@@ -48,6 +48,7 @@ func (s *IdentityService) CreateHeadlessAuthenticationStub(ctx context.Context,
 	if _, err = s.Create(ctx, *item); err != nil {
 		return nil, trace.Wrap(err)
 	}
+
 	return headlessAuthn, nil
 }
 
@@ -87,9 +88,31 @@ func (s *IdentityService) GetHeadlessAuthentication(ctx context.Context, name st
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+
 	return headlessAuthn, nil
 }
 
+// GetHeadlessAuthentications returns all headless authentications from the backend.
+func (s *IdentityService) GetHeadlessAuthentications(ctx context.Context) ([]*types.HeadlessAuthentication, error) {
+	rangeStart := headlessAuthenticationKey("")
+	rangeEnd := backend.RangeEnd(rangeStart)
+	items, err := s.GetRange(ctx, rangeStart, rangeEnd, 0)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	headlessAuthns := make([]*types.HeadlessAuthentication, len(items.Items))
+	for i, item := range items.Items {
+		headlessAuthn, err := unmarshalHeadlessAuthenticationFromItem(&item)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		headlessAuthns[i] = headlessAuthn
+	}
+
+	return headlessAuthns, nil
+}
+
 // DeleteHeadlessAuthentication deletes a headless authentication from the backend by name.
 func (s *IdentityService) DeleteHeadlessAuthentication(ctx context.Context, name string) error {
 	err := s.Delete(ctx, headlessAuthenticationKey(name))
@@ -119,7 +142,9 @@ func unmarshalHeadlessAuthenticationFromItem(item *backend.Item) (*types.Headles
 		return nil, trace.Wrap(err, "error unmarshalling headless authentication from storage")
 	}
 
-	headlessAuthn.Metadata.Expires = &item.Expires
+	// Copy item.Expires without pointer to avoid race conditions with memory backend.
+	headlessAuthn.Metadata.Expires = new(time.Time)
+	*headlessAuthn.Metadata.Expires = item.Expires
 	if err := headlessAuthn.CheckAndSetDefaults(); err != nil {
 		return nil, trace.Wrap(err)
 	}
diff --git a/lib/services/local/headlessauthn_test.go b/lib/services/local/headlessauthn_test.go
index b79a90f9ed382..a1e9b9dc1ef40 100644
--- a/lib/services/local/headlessauthn_test.go
+++ b/lib/services/local/headlessauthn_test.go
@@ -141,6 +141,10 @@ func TestIdentityService_HeadlessAuthenticationBackend(t *testing.T) {
 			retrieved, err := identity.GetHeadlessAuthentication(ctx, test.ha.Metadata.Name)
 			require.NoError(t, err, "GetHeadlessAuthentication returned non-nil error")
 			require.Equal(t, swapped, retrieved)
+
+			retrievedList, err := identity.GetHeadlessAuthentications(ctx)
+			require.NoError(t, err, "GetHeadlessAuthentications returned non-nil error")
+			require.Equal(t, []*types.HeadlessAuthentication{swapped}, retrievedList)
 		})
 	}
 }
diff --git a/lib/services/local/headlessauthn_watcher.go b/lib/services/local/headlessauthn_watcher.go
new file mode 100644
index 0000000000000..52a0e841b9d58
--- /dev/null
+++ b/lib/services/local/headlessauthn_watcher.go
@@ -0,0 +1,357 @@
+/*
+Copyright 2023 Gravitational, Inc.
+
+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 local
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
+	"github.com/sirupsen/logrus"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/utils/retryutils"
+	"github.com/gravitational/teleport/lib/backend"
+	"github.com/gravitational/teleport/lib/defaults"
+	"github.com/gravitational/teleport/lib/utils"
+)
+
+// maxWaiters is the maximum number of concurrent waiters that a headless authentication watcher
+// will accept. This limit is introduced because the headless login flow creates waiters from an
+// unauthenticated endpoint, which could be exploited in a ddos attack without the limit in place.
+//
+// 1024 was chosen as a reasonable limit, as under normal conditions, a single Teleport Cluster
+// would never have over 1000 concurrent headless logins, each of which has a maximum lifetime
+// of 30-60 seconds. If this limit is exceeded in a reasonable scenario, this limit should be
+// made configurable in the server configuration file.
+const maxWaiters = 1024
+
+var watcherClosedErr = trace.Errorf("headless authentication watcher closed")
+
+type HeadlessAuthenticationWatcherConfig struct {
+	// Backend is the storage backend used to create watchers.
+	Backend backend.Backend
+	// Log is a logger.
+	Log logrus.FieldLogger
+	// Clock is used to control time.
+	Clock clockwork.Clock
+	// MaxRetryPeriod is the maximum retry period on failed watchers.
+	MaxRetryPeriod time.Duration
+}
+
+// CheckAndSetDefaults checks parameters and sets default values.
+func (cfg *HeadlessAuthenticationWatcherConfig) CheckAndSetDefaults() error {
+	if cfg.Backend == nil {
+		return trace.BadParameter("missing parameter Backend")
+	}
+	if cfg.Log == nil {
+		cfg.Log = logrus.StandardLogger()
+		cfg.Log.WithField("resource-kind", types.KindHeadlessAuthentication)
+	}
+	if cfg.MaxRetryPeriod == 0 {
+		// On watcher failure, we eagerly retry in order to avoid login delays.
+		cfg.MaxRetryPeriod = defaults.HighResPollingPeriod
+	}
+	if cfg.Clock == nil {
+		cfg.Clock = cfg.Backend.Clock()
+	}
+	return nil
+}
+
+// HeadlessAuthenticationWatcher is a custom backend watcher for the headless authentication resource.
+type HeadlessAuthenticationWatcher struct {
+	HeadlessAuthenticationWatcherConfig
+	identityService *IdentityService
+	retry           retryutils.Retry
+	mux             sync.Mutex
+	waiters         [maxWaiters]headlessAuthenticationWaiter
+	closed          chan struct{}
+}
+
+// NewHeadlessAuthenticationWatcher creates a new headless authentication resource watcher.
+// The watcher will close once the given ctx is closed.
+func NewHeadlessAuthenticationWatcher(ctx context.Context, cfg HeadlessAuthenticationWatcherConfig) (*HeadlessAuthenticationWatcher, error) {
+	if err := cfg.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	retry, err := retryutils.NewLinear(retryutils.LinearConfig{
+		First:  utils.FullJitter(cfg.MaxRetryPeriod / 10),
+		Step:   cfg.MaxRetryPeriod / 5,
+		Max:    cfg.MaxRetryPeriod,
+		Jitter: retryutils.NewHalfJitter(),
+		Clock:  cfg.Clock,
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	watcher := &HeadlessAuthenticationWatcher{
+		HeadlessAuthenticationWatcherConfig: cfg,
+		identityService:                     NewIdentityService(cfg.Backend),
+		retry:                               retry,
+		closed:                              make(chan struct{}),
+	}
+
+	go watcher.runWatchLoop(ctx)
+
+	return watcher, nil
+}
+
+func (h *HeadlessAuthenticationWatcher) close() {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+	close(h.closed)
+}
+
+func (h *HeadlessAuthenticationWatcher) runWatchLoop(ctx context.Context) {
+	defer h.close()
+	for {
+		err := h.watch(ctx)
+
+		startedWaiting := h.Clock.Now()
+		select {
+		case t := <-h.retry.After():
+			h.Log.Debugf("Attempting to restart watch after waiting %v.", t.Sub(startedWaiting))
+			h.retry.Inc()
+		case <-ctx.Done():
+			h.Log.Debug("Context closed, returning from watch loop.")
+			return
+		case <-h.closed:
+			h.Log.Debug("Watcher closed, returning from watch loop.")
+			return
+		}
+		if err != nil {
+			h.Log.Warningf("Restart watch on error: %v.", err)
+		}
+	}
+}
+
+func (h *HeadlessAuthenticationWatcher) watch(ctx context.Context) error {
+	watcher, err := h.Backend.NewWatcher(ctx, backend.Watch{
+		Name:            types.KindHeadlessAuthentication,
+		MetricComponent: types.KindHeadlessAuthentication,
+		Prefixes:        [][]byte{headlessAuthenticationKey("")},
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	defer watcher.Close()
+
+	select {
+	case <-watcher.Done():
+		return fmt.Errorf("watcher closed")
+	case <-ctx.Done():
+		return ctx.Err()
+	case event := <-watcher.Events():
+		if event.Type != types.OpInit {
+			return trace.BadParameter("expected init event, got %v instead", event.Type)
+		}
+	}
+
+	headlessAuthns, err := h.identityService.GetHeadlessAuthentications(ctx)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	// Notify any waiters initiated before the new watcher initialized.
+	h.notify(headlessAuthns...)
+	h.retry.Reset()
+
+	for {
+		select {
+		case event := <-watcher.Events():
+			switch event.Type {
+			case types.OpPut:
+				headlessAuthn, err := unmarshalHeadlessAuthenticationFromItem(&event.Item)
+				if err != nil {
+					h.Log.WithError(err).Debug("failed to unmarshal headless authentication from put event")
+				} else {
+					h.notify(headlessAuthn)
+				}
+			}
+		case <-watcher.Done():
+			return fmt.Errorf("watcher closed")
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func (h *HeadlessAuthenticationWatcher) notify(headlessAuthns ...*types.HeadlessAuthentication) {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+	for _, headlessAuthn := range headlessAuthns {
+		for i := range h.waiters {
+			if h.waiters[i].name == headlessAuthn.Metadata.Name {
+				select {
+				case h.waiters[i].ch <- headlessAuthn:
+				default:
+					h.markStaleUnderLock(&h.waiters[i])
+				}
+			}
+		}
+	}
+}
+
+// CheckWaiter checks if there is an active waiter matching the given
+// headless authentication ID. Used in tests.
+func (h *HeadlessAuthenticationWatcher) CheckWaiter(name string) bool {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+	for i := range h.waiters {
+		if h.waiters[i].name == name {
+			return true
+		}
+	}
+	return false
+}
+
+// Wait watches for the headless authentication with the given id to be added/updated
+// in the backend, and waits for the given condition to be met, to result in an error,
+// or for the given context to close.
+func (h *HeadlessAuthenticationWatcher) Wait(ctx context.Context, name string, cond func(*types.HeadlessAuthentication) (bool, error)) (*types.HeadlessAuthentication, error) {
+	waiter, err := h.assignWaiter(ctx, name)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	defer h.unassignWaiter(waiter)
+
+	checkBackend := func() (*types.HeadlessAuthentication, bool, error) {
+		headlessAuthn, err := h.identityService.GetHeadlessAuthentication(ctx, name)
+		if err != nil {
+			return nil, false, trace.Wrap(err)
+		}
+
+		ok, err := cond(headlessAuthn)
+		if err != nil {
+			return nil, false, trace.Wrap(err)
+		}
+
+		return headlessAuthn, ok, nil
+	}
+
+	// With the waiter allocated, check if there is an existing entry in the backend.
+	headlessAuthn, ok, err := checkBackend()
+	if err != nil && !trace.IsNotFound(err) {
+		return nil, trace.Wrap(err)
+	} else if ok {
+		return headlessAuthn, nil
+	}
+
+	for {
+		select {
+		case <-waiter.stale:
+			// If the waiter is a slow consumer it may be marked as stale, in which
+			// case it should check the backend for the latest resource version.
+			h.unmarkStale(waiter)
+
+			headlessAuthn, ok, err := checkBackend()
+			if err != nil {
+				return nil, trace.Wrap(err)
+			} else if ok {
+				return headlessAuthn, nil
+			}
+		case headlessAuthn := <-waiter.ch:
+			select {
+			case <-waiter.stale:
+				// prioritize stale check.
+				continue
+			default:
+			}
+			if ok, err := cond(headlessAuthn); err != nil {
+				return nil, trace.Wrap(err)
+			} else if ok {
+				return headlessAuthn, nil
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-h.closed:
+			return nil, watcherClosedErr
+		}
+	}
+}
+
+func (h *HeadlessAuthenticationWatcher) assignWaiter(ctx context.Context, name string) (*headlessAuthenticationWaiter, error) {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+
+	select {
+	case <-h.closed:
+		return nil, watcherClosedErr
+	default:
+	}
+
+	for i := range h.waiters {
+		if h.waiters[i].ch != nil {
+			continue
+		}
+		h.waiters[i].ch = make(chan *types.HeadlessAuthentication)
+		h.waiters[i].name = name
+		h.waiters[i].stale = make(chan struct{})
+		return &h.waiters[i], nil
+	}
+
+	return nil, trace.LimitExceeded("too many in-flight headless login requests")
+}
+
+func (h *HeadlessAuthenticationWatcher) unassignWaiter(waiter *headlessAuthenticationWaiter) {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+
+	// close channels.
+	close(waiter.ch)
+	h.markStaleUnderLock(waiter)
+
+	waiter.ch = nil
+	waiter.name = ""
+	waiter.stale = nil
+}
+
+// headlessAuthenticationWaiter is a waiter for a specific headless authentication.
+type headlessAuthenticationWaiter struct {
+	// name is the name of the headless authentication resource being waited on.
+	name string
+	// ch is a channel used by the watcher to send resource updates.
+	ch chan *types.HeadlessAuthentication
+	// stale is a channel used to determine if the waiter is stale and
+	// needs to check the backend for missed data. The watcher will close
+	// this channel when it misses an update.
+	stale chan struct{}
+}
+
+// markStaleUnderLock marks a waiter as stale so it will update itself once available.
+// This should be called when a waiter misses an update due to slow consumption on its channel.
+//
+// must be called by HeadlessAuthenticationWatcher under watcherMux
+func (h *HeadlessAuthenticationWatcher) markStaleUnderLock(waiter *headlessAuthenticationWaiter) {
+	select {
+	case <-waiter.stale:
+	default:
+		close(waiter.stale)
+	}
+}
+
+// unmarkStale marks a waiter as not stale. This should be called when the waiter performs a stale check.
+func (h *HeadlessAuthenticationWatcher) unmarkStale(waiter *headlessAuthenticationWaiter) {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+	waiter.stale = make(chan struct{})
+}
diff --git a/lib/services/local/headlessauthn_watcher_test.go b/lib/services/local/headlessauthn_watcher_test.go
new file mode 100644
index 0000000000000..8338e37b7dd07
--- /dev/null
+++ b/lib/services/local/headlessauthn_watcher_test.go
@@ -0,0 +1,196 @@
+/*
+Copyright 2023 Gravitational, Inc.
+
+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 local_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/services"
+	"github.com/gravitational/teleport/lib/services/local"
+)
+
+func TestHeadlessAuthenticationWatcher(t *testing.T) {
+	ctx := context.Background()
+
+	t.Parallel()
+	identity := newIdentityService(t, clockwork.NewFakeClock())
+
+	watcherCtx, watcherCancel := context.WithCancel(ctx)
+	defer watcherCancel()
+
+	watcherClock := clockwork.NewFakeClock()
+	w, err := local.NewHeadlessAuthenticationWatcher(watcherCtx, local.HeadlessAuthenticationWatcherConfig{
+		Clock:   watcherClock,
+		Backend: identity.Backend,
+	})
+	require.NoError(t, err)
+
+	pubUUID := services.NewHeadlessAuthenticationID([]byte(sshPubKey))
+
+	waitInGoroutine := func(ctx context.Context, t *testing.T, name string, cond func(*types.HeadlessAuthentication) (bool, error)) (chan *types.HeadlessAuthentication, chan error) {
+		headlessAuthnCh := make(chan *types.HeadlessAuthentication, 1)
+		errC := make(chan error, 1)
+		go func() {
+			headlessAuthn, err := w.Wait(ctx, name, cond)
+			errC <- err
+			headlessAuthnCh <- headlessAuthn
+		}()
+		require.Eventually(t, func() bool { return w.CheckWaiter(name) }, time.Millisecond*100, time.Millisecond*10)
+
+		return headlessAuthnCh, errC
+	}
+
+	t.Run("WaitTimeout", func(t *testing.T) {
+		waitCtx, waitCancel := context.WithTimeout(ctx, time.Millisecond*10)
+		defer waitCancel()
+
+		_, err = w.Wait(waitCtx, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) { return true, nil })
+		require.Error(t, err)
+		require.Equal(t, waitCtx.Err(), err)
+	})
+
+	t.Run("WaitCreateStub", func(t *testing.T) {
+		waitCtx, waitCancel := context.WithTimeout(ctx, time.Millisecond*500)
+		defer waitCancel()
+
+		headlessAuthnCh, errC := waitInGoroutine(waitCtx, t, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) {
+			return true, nil
+		})
+
+		stub, err := identity.CreateHeadlessAuthenticationStub(ctx, pubUUID)
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, identity.DeleteHeadlessAuthentication(ctx, pubUUID)) })
+
+		require.NoError(t, <-errC)
+		require.Equal(t, stub, <-headlessAuthnCh)
+	})
+
+	t.Run("WaitCompareAndSwap", func(t *testing.T) {
+		stub, err := identity.CreateHeadlessAuthenticationStub(ctx, pubUUID)
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, identity.DeleteHeadlessAuthentication(ctx, pubUUID)) })
+
+		waitCtx, waitCancel := context.WithTimeout(ctx, time.Millisecond*500)
+		defer waitCancel()
+
+		headlessAuthnCh, errC := waitInGoroutine(waitCtx, t, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) {
+			return ha.State == types.HeadlessAuthenticationState_HEADLESS_AUTHENTICATION_STATE_APPROVED, nil
+		})
+
+		replace := *stub
+		replace.State = types.HeadlessAuthenticationState_HEADLESS_AUTHENTICATION_STATE_APPROVED
+		replace.PublicKey = []byte(sshPubKey)
+		replace.User = "user"
+
+		swapped, err := identity.CompareAndSwapHeadlessAuthentication(ctx, stub, &replace)
+		require.NoError(t, err)
+
+		require.NoError(t, <-errC)
+		require.Equal(t, swapped, <-headlessAuthnCh)
+	})
+
+	t.Run("StaleCheck", func(t *testing.T) {
+		waitCtx, waitCancel := context.WithTimeout(ctx, time.Millisecond*500)
+		defer waitCancel()
+
+		// Create two waiters - a blocked consumer and a free consumer.
+		blockWait := make(chan struct{})
+		_, errC := waitInGoroutine(waitCtx, t, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) {
+			<-blockWait
+			return false, nil
+		})
+
+		notifyReceived := make(chan struct{}, 1)
+		waitInGoroutine(waitCtx, t, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) {
+			notifyReceived <- struct{}{}
+			return false, nil
+		})
+
+		// Create stub and wait for it to be caught by the free waiter.
+		stub, err := identity.CreateHeadlessAuthenticationStub(ctx, pubUUID)
+		require.NoError(t, err)
+		<-notifyReceived
+
+		replace := *stub
+		replace.PublicKey = []byte(sshPubKey)
+		replace.User = "user"
+
+		// perform a put to mark the blocked waiter as stale and
+		// wait for it to be caught by the free waiter.
+		_, err = identity.CompareAndSwapHeadlessAuthentication(ctx, stub, &replace)
+		require.NoError(t, err)
+		<-notifyReceived
+
+		// delete the headless authentication and unblock.
+		err = identity.DeleteHeadlessAuthentication(ctx, pubUUID)
+		require.NoError(t, err)
+		close(blockWait)
+
+		// the blocked waiter should perform a stale check and return a not found error.
+		err = <-errC
+		require.True(t, trace.IsNotFound(err), "Expected a not found error from Wait but got %v", err)
+	})
+
+	t.Run("WatchReset", func(t *testing.T) {
+		waitCtx, waitCancel := context.WithTimeout(ctx, time.Millisecond*500)
+		defer waitCancel()
+
+		headlessAuthnCh, errC := waitInGoroutine(waitCtx, t, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) {
+			return true, nil
+		})
+
+		// closed watchers should be handled gracefully and reset.
+		identity.Backend.CloseWatchers()
+		watcherClock.BlockUntil(1)
+
+		// The watcher should notify waiters of missed events.
+		stub, err := identity.CreateHeadlessAuthenticationStub(ctx, pubUUID)
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, identity.DeleteHeadlessAuthentication(ctx, pubUUID)) })
+
+		watcherClock.Advance(w.MaxRetryPeriod)
+		require.NoError(t, <-errC)
+		require.Equal(t, stub, <-headlessAuthnCh)
+	})
+
+	t.Run("WatcherClosed", func(t *testing.T) {
+		waitCtx, waitCancel := context.WithTimeout(ctx, time.Millisecond*500)
+		defer waitCancel()
+
+		_, errC := waitInGoroutine(waitCtx, t, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) {
+			return true, nil
+		})
+
+		watcherCancel()
+
+		// waiters should be notified to close and result in ctx error
+		waitErr := <-errC
+		require.Error(t, waitErr)
+		require.Equal(t, waitErr.Error(), "headless authentication watcher closed")
+
+		// New waiters should be prevented.
+		_, err = w.Wait(ctx, pubUUID, func(ha *types.HeadlessAuthentication) (bool, error) { return true, nil })
+		require.Error(t, err)
+	})
+}