types, auth, noise: let one request wait on each auth session

One verdict per session, so one waiter: the newest. A displaced one
returns at once (register error, SSH re-delegate), not at session end.
This commit is contained in:
Kristoffer Dalby committed 2026-10-27 21:32:33 +00:00
1 parent 71150e7990
commit d2749d9726
5 files changed
+188 -2

No files matched your search

+8
View File
@@ -342,6 +342,8 @@ func (h *Headscale) waitForFollowup(
)
}
superseded := reg.Park()
var verdict types.AuthVerdict
select {
// Prefer a completed registration even if the context has also
@@ -353,6 +355,12 @@ func (h *Headscale) waitForFollowup(
select {
case <-ctx.Done():
return nil, NewHTTPError(http.StatusUnauthorized, "registration timed out", ctx.Err())
case <-superseded:
// tailscaled keeps one follow-up in flight, so only a request
// it already abandoned is displaced.
return nil, NewHTTPError(
http.StatusConflict, "superseded by a newer followup request", nil,
)
case verdict = <-reg.WaitForAuth():
}
}
+83
View File
@@ -4690,3 +4690,86 @@ func TestPendingRegistrationKeepsNoRoutes(t *testing.T) {
assert.Equal(t, want, entry.RegistrationData().Hostinfo)
assert.Len(t, hostinfo.RoutableIPs, 1, "the request itself must be left alone")
}
// TestFollowupParksOneRequestPerRegistration pins that a pending registration
// keeps one waiting follow-up, the newest. The session hands out one verdict,
// so older waiters are answered at once instead of staying parked until the
// session ends and then each opening a new one.
func TestFollowupParksOneRequestPerRegistration(t *testing.T) {
app := createTestApp(t)
machineKey := key.NewMachine().Public()
authID := types.MustAuthID()
reg := types.NewRegisterAuthRequest(&types.RegistrationData{
MachineKey: machineKey,
Hostname: "followup-park-node",
})
app.state.SetAuthCacheEntry(authID, reg)
req := tailcfg.RegisterRequest{
NodeKey: key.NewNode().Public(),
Followup: fmt.Sprintf("http://localhost:8080/register/%s", authID),
}
const followups = 50
type answer struct {
resp *tailcfg.RegisterResponse
err error
}
answered := make(chan answer, followups)
for range followups {
go func() {
resp, err := app.waitForFollowup(t.Context(), req, machineKey)
answered <- answer{resp, err}
}()
}
next := func(msg string) answer {
t.Helper()
select {
case a := <-answered:
return a
case <-time.After(5 * time.Second):
require.FailNow(t, msg)
return answer{}
}
}
for range followups - 1 {
var httpErr HTTPError
require.ErrorAs(t,
next("more than one follow-up is parked on the registration").err,
&httpErr)
assert.Equal(t, http.StatusConflict, httpErr.Code)
}
select {
case a := <-answered:
require.FailNowf(t, "the newest follow-up must still be waiting", "got %+v", a)
case <-time.After(50 * time.Millisecond):
}
cacheLen := func() int {
n, _ := app.state.DebugRegistrationCache()["current_len"].(int)
return n
}
before := cacheLen()
user := app.state.CreateUserForTest("followup-park-user")
node := app.state.CreateNodeForTest(user, "followup-park-node")
node.MachineKey = machineKey
node.User = user
reg.FinishAuth(types.AuthVerdict{Node: node.View()})
last := next("the parked follow-up did not collect the verdict")
require.NoError(t, last.err)
assert.True(t, last.resp.MachineAuthorized)
assert.LessOrEqual(t, cacheLen(), before+1,
"the verdict must not open a session per waiter")
}
+9 -2
View File
@@ -598,8 +598,9 @@ func (ns *noiseServer) sshActionHoldAndDelegate(
}
// sshActionFollowUp handles follow-up requests where the client
// provides an auth_id. It blocks until the auth session resolves or
// the request context is cancelled (e.g. the client disconnects).
// provides an auth_id. It blocks until the auth session resolves, the
// request context is cancelled (e.g. the client disconnects), or a newer
// follow-up parks on the session.
func (ns *noiseServer) sshActionFollowUp(
ctx context.Context,
reqLog zerolog.Logger,
@@ -663,6 +664,8 @@ func (ns *noiseServer) sshActionFollowUp(
reqLog.Trace().Caller().Msg("SSH action follow-up")
superseded := auth.Park()
var (
verdict types.AuthVerdict
verdictOK bool
@@ -680,6 +683,10 @@ func (ns *noiseServer) sshActionFollowUp(
"ssh action follow-up cancelled",
ctx.Err(),
)
case <-superseded:
// The newer request collects the verdict; this one re-delegates, as
// it would on a consumed verdict, without waiting for it.
return sessionGone("SSH check follow-up superseded; re-delegating")
case verdict, verdictOK = <-auth.WaitForAuth():
}
+68
View File
@@ -2041,3 +2041,71 @@ func TestNoiseRouter_DoesNotServeMetrics(t *testing.T) {
assert.Equal(t, http.StatusNotFound, resp.StatusCode)
}
// TestSSHActionFollowUp_ParksOneRequestPerSession pins that a pending SSH
// check keeps one waiting follow-up, the newest. Older waiters re-delegate at
// once instead of staying parked until the session ends.
func TestSSHActionFollowUp_ParksOneRequestPerSession(t *testing.T) {
t.Parallel()
app := createTestApp(t)
nodes := putSSHCheckNodes(t, app, "ssh-park-user", "src-node", "dst-node")
src, dst := nodes[0], nodes[1]
authID := types.MustAuthID()
auth := types.NewSSHCheckAuthRequest(src.ID, dst.ID)
app.state.SetAuthCacheEntry(authID, auth)
ns := &noiseServer{headscale: app, machineKey: dst.MachineKey}
const followups = 50
type answer struct {
action *tailcfg.SSHAction
err error
}
answered := make(chan answer, followups)
for range followups {
go func() {
action, err := ns.sshActionFollowUp(
t.Context(), zerolog.Nop(), &tailcfg.SSHAction{},
authID.String(), src.ID, dst.ID, sshTestLocalUser,
)
answered <- answer{action, err}
}()
}
next := func(msg string) answer {
t.Helper()
select {
case a := <-answered:
return a
case <-time.After(5 * time.Second):
require.FailNow(t, msg)
return answer{}
}
}
for range followups - 1 {
a := next("more than one follow-up is parked on the session")
require.NoError(t, a.err)
assert.NotEqual(t, authID, requireSSHHold(t, *a.action),
"re-delegation must mint a new session")
}
select {
case a := <-answered:
require.FailNowf(t, "the newest follow-up must still be waiting", "got %+v", a)
case <-time.After(50 * time.Millisecond):
}
auth.FinishAuth(types.AuthVerdict{})
last := next("the parked follow-up did not collect the verdict")
require.NoError(t, last.err)
assert.True(t, last.action.Accept, "accept, got %+v", last.action)
}
+20
View File
@@ -143,6 +143,10 @@ type AuthRequest struct {
finished chan AuthVerdict
closed *atomic.Bool
// waiter is closed once a newer request parks on this session; see
// [AuthRequest.Park].
waiter atomic.Pointer[chan struct{}]
}
// NewAuthRequest creates a pending auth request with no payload, suitable
@@ -245,6 +249,22 @@ func (rn *AuthRequest) WaitForAuth() <-chan AuthVerdict {
return rn.finished
}
// Park makes the caller the one request waiting on [AuthRequest.WaitForAuth]
// and returns a channel that is closed once a newer caller takes its place.
// [AuthRequest.FinishAuth] hands out one verdict, so a session needs one
// waiter. The newest wins, so a client that reconnected is not kept out by
// its own dead request. Parked requests are then bounded by the sessions in
// the auth cache, not by how many requests clients send.
func (rn *AuthRequest) Park() <-chan struct{} {
waiter := make(chan struct{})
if prev := rn.waiter.Swap(&waiter); prev != nil {
close(*prev)
}
return waiter
}
type AuthVerdict struct {
// Err is the error that occurred during the authentication process, if any.
// If Err is nil, the authentication process has succeeded.