Share PBS session renewal across concurrent monitoring reads

Publish ticket, CSRF and expiry together and share only an expired session renewal. Preserve independent requests, context cancellation without false credential errors, JSON/form authentication, token authority and terminal API refusals without replay. Add registered pinned HTTPS race, failure and recovery controls plus a canonical consuming projection. Keep exact verification paths sorted as required by candidate admission; runtime and tests are unchanged from the source-proven candidate.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,27 @@
 
 ## Purpose
 
+### PBS password renewal publishes one complete shared session
+
+Concurrent reads of an expired PBS password session share one in-flight
+authentication attempt. Ticket, CSRF token and expiry are published together;
+each ordinary request uses one complete tuple. Only renewal is shared: the
+subsequent API reads remain independent, and a cancelled waiter cannot cancel
+the owner. Caller cancellation and deadlines stay context errors, not credential
+refusals, including token reads that never attempt a login. Joined callers receive the owner's failure without reissuing it;
+a later ordinary call may try again. JSON authentication and the existing
+400/415 form fallback retain their semantics. Tokens do not acquire session
+authority, and API 401/403 replies remain terminal without adding a replay.
+
+`pkg/pbs/auth_session_concurrency_test.go` exercises pinned local HTTPS expiry,
+encoding fallback/refusal, malformed responses, independent reads, cookie/CSRF
+pairing, owner/waiter cancellation and ordinary later recovery. Its accepted
+node/datastore responses also consume through the canonical PBS projection.
+TLS pins/CA/opt-out and redirect policy, credentials, ticket lifetime, provider
+permissions, wire shapes, polling schedules, guest/backup fences and stored
+History are unchanged. These synthetic controls do not establish an installed
+connection, native recovery or the cause of a reported PBS/guest failure.
+
 ### Connection retirement preserves unrelated guest safety state
 
 Retiring a consolidated PVE connection cancels and clears only its own guest

docs/release-control/v6/internal/subsystems/registry.json

@@ -6466,6 +6466,7 @@
             "allow_same_subsystem_tests": false,
             "test_prefixes": [],
             "exact_files": [
+              "pkg/pbs/auth_session_concurrency_test.go",
               "pkg/pbs/client_http_test.go",
               "pkg/pbs/client_test.go"
             ]

pkg/pbs/auth_session_concurrency_test.go

@@ -0,0 +1,481 @@
+package pbs
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	monitorerrors "github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors"
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+)
+
+// These are pinned, guest-local HTTPS fixtures with synthetic credentials.
+// No backup, guest-agent, production authentication or installed recovery runs.
+type pbsSessionFixture struct {
+	client                                                *Client
+	ctx                                                   context.Context
+	entered, finish, readsArrived                         chan struct{}
+	finishOnce                                            sync.Once
+	authCalls, jsonCalls, formCalls, apiCalls, mismatched atomic.Int32
+	outcome                                               atomic.Value
+	gateReads                                             atomic.Bool
+	phase                                                 atomic.Int32
+}
+
+// Done observes arrival at a pending wait (or the old HTTP transport). It does
+// not inspect production-only counters, sleep or choose a favourable repeat.
+type pbsSessionJoinContext struct {
+	context.Context
+	joined chan struct{}
+	once   sync.Once
+}
+
+func (c *pbsSessionJoinContext) Done() <-chan struct{} {
+	c.once.Do(func() { close(c.joined) })
+	return c.Context.Done()
+}
+func pbsSessionAwait(t *testing.T, ch <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(5 * time.Second):
+		t.Fatal("PBS session fixture barrier did not complete")
+	}
+}
+func (f *pbsSessionFixture) release() { f.finishOnce.Do(func() { close(f.finish) }) }
+func newPBSSessionFixture(t *testing.T, outcome string, gateReads bool) *pbsSessionFixture {
+	t.Helper()
+	f := &pbsSessionFixture{entered: make(chan struct{}), finish: make(chan struct{}), readsArrived: make(chan struct{})}
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	f.ctx = ctx
+	f.outcome.Store(outcome)
+	f.gateReads.Store(gateReads)
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/access/ticket") {
+			n := f.authCalls.Add(1)
+			form := r.Header.Get("Content-Type") == "application/x-www-form-urlencoded"
+			if form {
+				f.formCalls.Add(1)
+			} else {
+				f.jsonCalls.Add(1)
+			}
+			if n > 1 {
+				if n == 2 {
+					close(f.entered)
+				}
+				select {
+				case <-f.finish:
+				case <-ctx.Done():
+					return
+				}
+				mode := f.outcome.Load().(string)
+				if !form && (mode == "form-400" || mode == "form-415") {
+					status := 400
+					if mode == "form-415" {
+						status = 415
+					}
+					w.WriteHeader(status)
+					fmt.Fprint(w, "synthetic encoding refusal")
+					return
+				}
+				switch mode {
+				case "refused-401":
+					w.WriteHeader(401)
+					fmt.Fprint(w, "synthetic renewal refusal")
+					return
+				case "refused-403":
+					w.WriteHeader(403)
+					fmt.Fprint(w, "synthetic renewal refusal")
+					return
+				case "unavailable":
+					w.WriteHeader(503)
+					fmt.Fprint(w, "synthetic renewal unavailable")
+					return
+				case "malformed":
+					fmt.Fprint(w, `{"data":`)
+					return
+				}
+			}
+			fmt.Fprintf(w, `{"data":{"ticket":"session-%d","CSRFPreventionToken":"csrf-%d"}}`, n, n)
+			return
+		}
+		cookie, err := r.Cookie("PBSAuthCookie")
+		if err != nil || !strings.HasPrefix(cookie.Value, "session-") {
+			f.mismatched.Add(1)
+			w.WriteHeader(401)
+			return
+		}
+		if r.Method != http.MethodGet && r.Header.Get("CSRFPreventionToken") != "csrf-"+strings.TrimPrefix(cookie.Value, "session-") {
+			f.mismatched.Add(1)
+			w.WriteHeader(403)
+			return
+		}
+		n := f.apiCalls.Add(1)
+		if f.gateReads.Load() {
+			if n == 8 {
+				close(f.readsArrived)
+			}
+			select {
+			case <-f.readsArrived:
+			case <-ctx.Done():
+				return
+			}
+		}
+		switch {
+		case strings.HasSuffix(r.URL.Path, "/version"):
+			fmt.Fprint(w, `{"data":{"version":"3.4.2","release":"1"}}`)
+		case strings.HasSuffix(r.URL.Path, "/admin/datastore"):
+			fmt.Fprint(w, `{"data":[{"store":"main"},{"store":"archive"}]}`)
+		case strings.HasSuffix(r.URL.Path, "/rrd"):
+			fmt.Fprint(w, `{"data":[]}`)
+		case strings.HasSuffix(r.URL.Path, "/gc"):
+			fmt.Fprint(w, `{"data":{}}`)
+		case strings.Contains(r.URL.Path, "/admin/datastore/"):
+			used := 250000000 + int64(f.phase.Load())*50000000
+			if strings.Contains(r.URL.Path, "/archive/") {
+				used = 800000000
+			}
+			fmt.Fprintf(w, `{"data":{"total":1000000000,"used":%d,"avail":%d}}`, used, 1000000000-used)
+		default:
+			used := int64(2147483648) + int64(f.phase.Load())*1073741824
+			cpu := 0.25 + float64(f.phase.Load())*0.25
+			fmt.Fprintf(w, `{"data":{"cpu":%f,"memory":{"total":8589934592,"used":%d,"free":%d},"uptime":%d}}`, cpu, used, 8589934592-used, 7200+f.phase.Load()*60)
+		}
+	}))
+	pin := sha256.Sum256(server.Certificate().Raw)
+	c, err := NewClient(ClientConfig{Host: server.URL, User: "synthetic@pam", Password: "synthetic-password", VerifySSL: true, Fingerprint: hex.EncodeToString(pin[:]), Timeout: 5 * time.Second})
+	if err != nil {
+		cancel()
+		server.Close()
+		t.Fatal(err)
+	}
+	f.client = c
+	// No concurrent callers exist while the fixture expires the initial ticket.
+	c.auth.expiresAt = time.Now().Add(-time.Hour)
+	t.Cleanup(func() { cancel(); f.release(); c.httpClient.CloseIdleConnections(); server.Close() })
+	return f
+}
+func (f *pbsSessionFixture) node(ctx context.Context) error {
+	status, err := f.client.GetNodeStatus(ctx)
+	if err == nil && (status == nil || status.CPU != 0.25 || status.Memory.Total != 8589934592 || status.Memory.Used != 2147483648 || status.Uptime != 7200) {
+		return fmt.Errorf("ordinary renewed node read lost its CPU/memory/uptime tuple")
+	}
+	return err
+}
+
+func TestPBSClientSessionConcurrentExpiry(t *testing.T) {
+	for _, mode := range []string{"json", "form-400", "form-415", "refused-401", "refused-403", "unavailable", "malformed"} {
+		t.Run(mode, func(t *testing.T) {
+			successful := mode == "json" || strings.HasPrefix(mode, "form-")
+			f := newPBSSessionFixture(t, mode, successful)
+			original := f.client.auth
+			done := make(chan error, 8)
+			go func() { done <- f.node(f.ctx) }()
+			pbsSessionAwait(t, f.entered)
+			for i := 1; i < 8; i++ {
+				joined := &pbsSessionJoinContext{Context: f.ctx, joined: make(chan struct{})}
+				go func() { done <- f.node(joined) }()
+				pbsSessionAwait(t, joined.joined)
+			}
+			// A pending renewal cannot admit reads or another password request.
+			if f.apiCalls.Load() != 0 || f.authCalls.Load() != 2 {
+				t.Errorf("pending auth/API=%d/%d, want 2/0", f.authCalls.Load(), f.apiCalls.Load())
+			}
+			f.release()
+			for i := 0; i < 8; i++ {
+				select {
+				case err := <-done:
+					if successful && err != nil {
+						t.Error(err)
+					}
+					if !successful && err == nil {
+						t.Error("failed renewal admitted an ordinary read")
+					}
+				case <-f.ctx.Done():
+					t.Fatal("concurrent node reads did not complete independently")
+				}
+			}
+			want := int32(2)
+			if strings.HasPrefix(mode, "form-") {
+				want = 3
+			}
+			if f.authCalls.Load() != want || f.mismatched.Load() != 0 {
+				t.Errorf("renewal/pair mismatch=%d/%d, want %d/0", f.authCalls.Load(), f.mismatched.Load(), want)
+			}
+			if f.jsonCalls.Load() != 2 || f.formCalls.Load() != want-2 {
+				t.Errorf("JSON/form=%d/%d, want 2/%d", f.jsonCalls.Load(), f.formCalls.Load(), want-2)
+			}
+			f.gateReads.Store(false)
+			if successful {
+				if f.apiCalls.Load() != 8 {
+					t.Errorf("ordinary reads=%d, want 8", f.apiCalls.Load())
+				}
+				// The same request producer pairs cookie and CSRF on the existing write
+				// path; this targets only a synthetic endpoint, not any PBS operation.
+				resp, err := f.client.post(f.ctx, "/synthetic-read", url.Values{"fixture": {"preserved"}})
+				if err != nil {
+					t.Fatal(err)
+				}
+				resp.Body.Close()
+			} else {
+				if f.client.auth != original || f.apiCalls.Load() != 0 {
+					t.Error("failed renewal changed the prior tuple or admitted a read")
+				}
+				f.outcome.Store("json")
+				if err := f.node(f.ctx); err != nil {
+					t.Fatal(err)
+				}
+				if f.authCalls.Load() != 3 || f.apiCalls.Load() != 1 {
+					t.Errorf("later ordinary recovery auth/API=%d/%d, want 3/1", f.authCalls.Load(), f.apiCalls.Load())
+				}
+			}
+			t.Logf("pinned HTTPS: renewal=%s; eight callers, one shared attempt; existing encoding/failure semantics retained", mode)
+		})
+	}
+}
+
+func TestPBSClientSessionWaiterCancellation(t *testing.T) {
+	f := newPBSSessionFixture(t, "json", false)
+	owner := make(chan error, 1)
+	go func() { owner <- f.node(f.ctx) }()
+	pbsSessionAwait(t, f.entered)
+	ctx, cancel := context.WithCancel(f.ctx)
+	defer cancel()
+	joined := &pbsSessionJoinContext{Context: ctx, joined: make(chan struct{})}
+	waiter := make(chan error, 1)
+	go func() { waiter <- f.node(joined) }()
+	pbsSessionAwait(t, joined.joined)
+	cancel()
+	select {
+	case err := <-waiter:
+		if !errors.Is(err, context.Canceled) || monitorerrors.IsAuthError(err) {
+			t.Errorf("waiter result: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("cancelled waiter remained blocked")
+	}
+	if f.authCalls.Load() != 2 || f.apiCalls.Load() != 0 {
+		t.Error("waiter replayed authentication or admitted a read")
+	}
+	select {
+	case err := <-owner:
+		t.Fatalf("waiter cancelled the owner: %v", err)
+	default:
+	}
+	f.release()
+	if err := <-owner; err != nil {
+		t.Fatal(err)
+	}
+	if err := f.node(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	if f.authCalls.Load() != 2 || f.apiCalls.Load() != 2 {
+		t.Errorf("owner/later auth/API=%d/%d, want 2/2", f.authCalls.Load(), f.apiCalls.Load())
+	}
+}
+
+func TestPBSClientSessionOwnerCancellation(t *testing.T) {
+	f := newPBSSessionFixture(t, "json", false)
+	ctx, cancel := context.WithCancel(f.ctx)
+	defer cancel()
+	done := make(chan error, 2)
+	go func() { done <- f.node(ctx) }()
+	pbsSessionAwait(t, f.entered)
+	joined := &pbsSessionJoinContext{Context: f.ctx, joined: make(chan struct{})}
+	go func() { done <- f.node(joined) }()
+	pbsSessionAwait(t, joined.joined)
+	cancel()
+	for i := 0; i < 2; i++ {
+		select {
+		case err := <-done:
+			if !errors.Is(err, context.Canceled) || monitorerrors.IsAuthError(err) {
+				t.Errorf("shared owner cancellation: %v", err)
+			}
+		case <-time.After(time.Second):
+			t.Fatal("joined caller lost cancelled renewal result")
+		}
+	}
+	if f.authCalls.Load() != 2 || f.apiCalls.Load() != 0 {
+		t.Error("cancelled renewal admitted work or another attempt")
+	}
+	f.release()
+	if err := f.node(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	if f.authCalls.Load() != 3 || f.apiCalls.Load() != 1 {
+		t.Errorf("later recovery auth/API=%d/%d, want 3/1", f.authCalls.Load(), f.apiCalls.Load())
+	}
+}
+
+func TestPBSClientSessionTuplePublication(t *testing.T) {
+	f := newPBSSessionFixture(t, "json", false)
+	f.release()
+	if err := f.node(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 2)
+	go func() {
+		for i := 0; i < 32; i++ {
+			if err := f.client.authenticate(f.ctx); err != nil {
+				done <- err
+				return
+			}
+		}
+		done <- nil
+	}()
+	go func() {
+		for i := 0; i < 64; i++ {
+			resp, err := f.client.post(f.ctx, "/synthetic-read", url.Values{"fixture": {"preserved"}})
+			if err != nil {
+				done <- err
+				return
+			}
+			resp.Body.Close()
+		}
+		done <- nil
+	}()
+	for i := 0; i < 2; i++ {
+		if err := <-done; err != nil {
+			t.Error(err)
+		}
+	}
+	if f.mismatched.Load() != 0 || f.apiCalls.Load() != 65 {
+		t.Errorf("pair mismatches/API=%d/%d, want 0/65", f.mismatched.Load(), f.apiCalls.Load())
+	}
+}
+
+func TestPBSClientSessionTokenAndAPIRefusals(t *testing.T) {
+	for _, mode := range []string{"token", "session"} {
+		for _, status := range []int{http.StatusUnauthorized, http.StatusForbidden} {
+			t.Run(fmt.Sprintf("%s/%d", mode, status), func(t *testing.T) {
+				var authCalls, apiCalls atomic.Int32
+				server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					if strings.HasSuffix(r.URL.Path, "/access/ticket") {
+						authCalls.Add(1)
+						fmt.Fprint(w, `{"data":{"ticket":"initial","CSRFPreventionToken":"initial-csrf"}}`)
+						return
+					}
+					apiCalls.Add(1)
+					if mode == "token" && (r.Header.Get("Authorization") != "PBSAPIToken=synthetic@pbs!pulse:synthetic-token" || r.Header.Get("Cookie") != "" || r.Header.Get("CSRFPreventionToken") != "") {
+						t.Error("token acquired session authority")
+					}
+					w.WriteHeader(status)
+					fmt.Fprint(w, "synthetic API refusal")
+				}))
+				defer server.Close()
+				pin := sha256.Sum256(server.Certificate().Raw)
+				cfg := ClientConfig{Host: server.URL, User: "synthetic@pam", Password: "synthetic-password", VerifySSL: true, Fingerprint: hex.EncodeToString(pin[:]), Timeout: time.Second}
+				if mode == "token" {
+					cfg.TokenName = "synthetic@pbs!pulse"
+					cfg.TokenValue = "synthetic-token"
+				}
+				c, err := NewClient(cfg)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer c.httpClient.CloseIdleConnections()
+				_, err = c.GetVersion(context.Background())
+				if got, known := HTTPStatus(err); !known || got != status {
+					t.Errorf("typed refusal lost: %v", err)
+				}
+				want := int32(0)
+				if mode == "session" {
+					want = 1
+				}
+				if authCalls.Load() != want || apiCalls.Load() != 1 {
+					t.Errorf("API refusal replayed: auth/API=%d/%d, want %d/1", authCalls.Load(), apiCalls.Load(), want)
+				}
+			})
+		}
+	}
+}
+
+// The actual client/decoder feeds the existing canonical PBS projection. The
+// node/datastore-to-model assignment is the unchanged monitor projection; this
+// is not an installed poll, stored History or whole-application acceptance.
+func TestPBSClientSessionCanonicalProducer(t *testing.T) {
+	f := newPBSSessionFixture(t, "json", false)
+	f.release()
+	project := func() unifiedresources.Resource {
+		status, err := f.client.GetNodeStatus(f.ctx)
+		if err != nil || status == nil {
+			t.Fatalf("node observation: %v", err)
+		}
+		version, err := f.client.GetVersion(f.ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		stores, err := f.client.GetDatastores(f.ctx)
+		if err != nil || len(stores) != 2 {
+			t.Fatalf("datastore observations: %d, %v", len(stores), err)
+		}
+		instance := models.PBSInstance{ID: "pbs-session", Name: "pbs-session", Host: "https://pbs-session.example.invalid:8007", Status: "online", ConnectionHealth: "healthy", Version: version.Version, CPU: status.CPU, Memory: float64(status.Memory.Used) / float64(status.Memory.Total) * 100, MemoryUsed: status.Memory.Used, MemoryTotal: status.Memory.Total, Uptime: status.Uptime, LastSeen: time.Now()}
+		for _, ds := range stores {
+			instance.Datastores = append(instance.Datastores, models.PBSDatastore{Name: ds.Store, Total: ds.Total, Used: ds.Used, Free: ds.Avail, Usage: float64(ds.Used) / float64(ds.Total) * 100, Status: ds.Status})
+		}
+		registry := unifiedresources.NewRegistry(nil)
+		registry.IngestSnapshot(models.StateSnapshot{PBSInstances: []models.PBSInstance{instance}})
+		resources := registry.ListByType(unifiedresources.ResourceTypePBS)
+		if len(resources) != 1 {
+			t.Fatalf("canonical PBS count=%d", len(resources))
+		}
+		return resources[0]
+	}
+	before := project()
+	f.phase.Store(1)
+	f.client.auth.expiresAt = time.Now().Add(-time.Hour) // No concurrent reader.
+	after := project()
+	if f.authCalls.Load() != 3 || f.mismatched.Load() != 0 {
+		t.Fatalf("ordinary producer renewal/pair mismatch = %d/%d", f.authCalls.Load(), f.mismatched.Load())
+	}
+	encoded, err := json.Marshal(map[string]unifiedresources.Resource{"before": before, "after": after})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("PULSE_PBS_SESSION_PRODUCER=" + string(encoded))
+}
+
+// A caller's exhausted budget is not a credential refusal. In particular,
+// token reads must not acquire a misleading authentication-failed wrapper.
+func TestPBSClientSessionCancelledReadKeepsErrorClass(t *testing.T) {
+	for _, mode := range []string{"token", "fresh-session", "expired-session"} {
+		t.Run(mode, func(t *testing.T) {
+			f := newPBSSessionFixture(t, "json", false)
+			c := f.client
+			if mode == "token" {
+				cfg := c.config
+				cfg.TokenName = "synthetic@pbs!pulse"
+				cfg.TokenValue = "synthetic-token"
+				var err error
+				c, err = NewClient(cfg)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer c.httpClient.CloseIdleConnections()
+			} else if mode == "fresh-session" {
+				c.auth.expiresAt = time.Now().Add(time.Hour)
+			}
+			ctx, cancel := context.WithCancel(f.ctx)
+			cancel()
+			_, err := c.GetVersion(ctx)
+			if !errors.Is(err, context.Canceled) || monitorerrors.IsAuthError(err) {
+				t.Errorf("unstarted/cancelled read acquired authentication failure: %v", err)
+			}
+			if f.authCalls.Load() != 1 || f.apiCalls.Load() != 0 {
+				t.Errorf("cancelled read admitted auth/API=%d/%d, want initial only/0", f.authCalls.Load(), f.apiCalls.Load())
+			}
+		})
+	}
+}

pkg/pbs/client.go

@@ -39,6 +39,11 @@ type Client struct {
 	auth       auth
 	config     ClientConfig
 
+	// Polling and datastore reads share this client. Publish a whole session
+	// tuple and share its renewal, not the subsequent requests or cancellation.
+	authMu       sync.Mutex
+	authInFlight *authRefresh
+
 	// GetNodeName caching. The node hostname is stable for the lifetime of a
 	// connection. PBS exposes /nodes as a superuser-only compatibility endpoint,
 	// so GetNodeName rejects API-token and non-superuser sessions before I/O.
@@ -54,6 +59,17 @@ type nodeNameAttempt struct {
 	err    error
 }
 
+type authRefresh struct {
+	done chan struct{}
+	err  error
+}
+
+func (c *Client) authSnapshot() auth {
+	c.authMu.Lock()
+	defer c.authMu.Unlock()
+	return c.auth
+}
+
 // ErrNodeNameUnavailableForAuth means the current PBS authentication mode
 // cannot access the superuser-only /nodes compatibility endpoint. Callers
 // should treat this as absent optional identity evidence, not as a connection
@@ -197,16 +213,47 @@ func NewClient(cfg ClientConfig) (*Client, error) {
 
 // authenticate performs password-based authentication
 func (c *Client) authenticate(ctx context.Context) error {
-	username := c.auth.user + "@" + c.auth.realm
-	password := c.config.Password
+	return c.refreshAuthentication(ctx, true)
+}
 
-	if err := c.authenticateJSON(ctx, username, password); err == nil {
-		return nil
-	} else if shouldFallbackToForm(err) {
-		return c.authenticateForm(ctx, username, password)
-	} else {
+// Ordinary expiry is checked under the same lock as session publication, so a
+// reader that observed an old ticket cannot start a second renewal. Explicit
+// authentication keeps its existing forced-refresh semantics. A joining caller
+// can stop waiting without cancelling the owner or replaying a failed attempt.
+func (c *Client) refreshAuthentication(ctx context.Context, force bool) error {
+	if err := ctx.Err(); err != nil {
 		return err
 	}
+	c.authMu.Lock()
+	if !force && (c.config.Password == "" || c.auth.tokenName != "" || !time.Now().After(c.auth.expiresAt)) {
+		c.authMu.Unlock()
+		return nil
+	}
+	if pending := c.authInFlight; pending != nil {
+		c.authMu.Unlock()
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-pending.done:
+			return pending.err
+		}
+	}
+	pending := &authRefresh{done: make(chan struct{})}
+	c.authInFlight = pending
+	username := c.auth.user + "@" + c.auth.realm
+	c.authMu.Unlock()
+	password := c.config.Password
+
+	err := c.authenticateJSON(ctx, username, password)
+	if shouldFallbackToForm(err) {
+		err = c.authenticateForm(ctx, username, password)
+	}
+	c.authMu.Lock()
+	pending.err = err
+	c.authInFlight = nil
+	close(pending.done)
+	c.authMu.Unlock()
+	return err
 }
 
 func (c *Client) authenticateJSON(ctx context.Context, username, password string) error {
@@ -287,9 +334,11 @@ func (c *Client) handleAuthResponse(resp *http.Response) error {
 		return err
 	}
 
+	c.authMu.Lock()
 	c.auth.ticket = result.Data.Ticket
 	c.auth.csrfToken = result.Data.CSRFPreventionToken
 	c.auth.expiresAt = time.Now().Add(2 * time.Hour) // PBS tickets expire after 2 hours
+	c.authMu.Unlock()
 
 	return nil
 }
@@ -360,11 +409,15 @@ func isPBSNotFoundError(err error) bool {
 // request performs an API request
 func (c *Client) request(ctx context.Context, method, path string, data url.Values) (*http.Response, error) {
 	// Re-authenticate if needed
-	if c.config.Password != "" && c.auth.tokenName == "" && time.Now().After(c.auth.expiresAt) {
-		if err := c.authenticate(ctx); err != nil {
-			return nil, fmt.Errorf("re-authentication failed: %w", err)
+	if err := c.refreshAuthentication(ctx, false); err != nil {
+		// An ended caller or cancelled renewal is not a credential refusal.
+		// Preserve its context error class, including token reads with no login.
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			return nil, err
 		}
+		return nil, fmt.Errorf("re-authentication failed: %w", err)
 	}
+	auth := c.authSnapshot()
 
 	var body io.Reader
 	if data != nil {
@@ -386,29 +439,29 @@ func (c *Client) request(ctx context.Context, method, path string, data url.Valu
 	}
 
 	// Set authentication
-	if c.auth.tokenName != "" && c.auth.tokenValue != "" {
+	if auth.tokenName != "" && auth.tokenValue != "" {
 		// API token authentication
 		// Note: tokenName already contains just the token part (e.g., "pulse-token")
 		// after parsing in NewClient, so we reconstruct the full format
 		authHeader := fmt.Sprintf("PBSAPIToken=%s@%s!%s:%s",
-			c.auth.user, c.auth.realm, c.auth.tokenName, c.auth.tokenValue)
+			auth.user, auth.realm, auth.tokenName, auth.tokenValue)
 		req.Header.Set("Authorization", authHeader)
 		// NEVER log the actual token value - only log that we're using token auth
 		// Log the auth header format (without the secret)
 		maskedHeader := fmt.Sprintf("PBSAPIToken=%s@%s!%s:***",
-			c.auth.user, c.auth.realm, c.auth.tokenName)
+			auth.user, auth.realm, auth.tokenName)
 		log.Debug().
-			Str("user", c.auth.user).
-			Str("realm", c.auth.realm).
-			Str("tokenName", c.auth.tokenName).
+			Str("user", auth.user).
+			Str("realm", auth.realm).
+			Str("tokenName", auth.tokenName).
 			Str("authHeaderFormat", maskedHeader).
 			Str("url", req.URL.String()).
 			Msg("Setting PBS API token authentication")
-	} else if c.auth.ticket != "" {
+	} else if auth.ticket != "" {
 		// Ticket authentication
-		req.Header.Set("Cookie", "PBSAuthCookie="+c.auth.ticket)
-		if method != "GET" && c.auth.csrfToken != "" {
-			req.Header.Set("CSRFPreventionToken", c.auth.csrfToken)
+		req.Header.Set("Cookie", "PBSAuthCookie="+auth.ticket)
+		if method != "GET" && auth.csrfToken != "" {
+			req.Header.Set("CSRFPreventionToken", auth.csrfToken)
 		}
 	}