Serialize Google connect and disconnect; consume connect state once

A disconnect read the stored refresh token, revoked it at Google, then
deleted the row only if it still held that token. Google revokes the whole
grant, so a reconnect whose token was issued before that revoke was kept
as connected with a dead token. Connect's exchange-and-store and
disconnect's read-revoke-delete now hold one mutex across the Google
request and the database write, and the delete is unconditional (the
conditional delete and ErrChanged are gone: the lock is the one guard).

The connect cookie was single-use only in that the browser was asked to
drop it; a saved copy verified again until it expired. The service now
keeps each started connect's state in memory until the cookie expires and
the callback consumes it, so a replayed cookie is refused as bad state.

The fake Google now issues a distinct refresh token per exchange, revokes
the whole grant like Google, and can hold a token or revoke request so
tests can interleave a reconnect and a disconnect.

Maestro task 1e9d7850-3a09-4572-b8a1-cb18cd4774cb (MQ125-01, MQ125-02).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/ARCHITECTURE.md

@@ -627,7 +627,10 @@ Runs are recorded in `schedule_runs` and executed through `jobs`.
     `access_type=offline`, `prompt=consent` and PKCE. Its `state` travels in
     the signed `__Host-mealquest_google` flow cookie (the same
     `auth.FlowCookie` mechanism as the login cookie), bound to the session by
-    a hash of the session's CSRF token.
+    a hash of the session's CSRF token. The service also keeps each started
+    connect's `state` in memory until the cookie expires; the callback
+    consumes it, so a replayed cookie is refused (`invalid_state`), and a
+    restart mid-connect means connecting again.
   - `GET /api/v1/integrations/google/callback` (the redirect URI:
     `MEALQUEST_PUBLIC_URL`'s origin plus this path) checks the state and
     session, exchanges the code, refuses a response with no refresh token
@@ -638,6 +641,10 @@ Runs are recorded in `schedule_runs` and executed through `jobs`.
   - `DELETE /api/v1/integrations/google` (CSRF-checked) revokes the token at
     Google, then deletes the row; a token Google reports as `invalid_token`
     counts as revoked, and any other revoke failure keeps the connection.
+  - Connect's exchange-and-store and disconnect's read-revoke-delete hold
+    one lock across the Google request and the database write: Google
+    revokes the whole grant, so a revoke overlapping a reconnect would
+    otherwise leave a stored token Google has already revoked.
   - No response, log line or error ever holds a token or Google's response
     text.
 - **Deterministic event ids**; each plan day records its

internal/auth/cookie.go

@@ -115,8 +115,9 @@ func (c FlowCookie) Set(w http.ResponseWriter, v flowValue, now time.Time) error
 	return nil
 }
 
-// Clear removes the cookie. A callback clears it before reading, so it is
-// single-use.
+// Clear asks the browser to discard the cookie. A copy of it still verifies
+// until it expires, so a flow that must be single-use also consumes its
+// state on the server.
 func (c FlowCookie) Clear(w http.ResponseWriter) {
 	http.SetCookie(w, &http.Cookie{
 		Name:     c.name,

internal/db/querier.go

@@ -50,10 +50,7 @@ type Querier interface {
 	// decided is left as it is (0 rows).
 	DecideSuggestion(ctx context.Context, arg DecideSuggestionParams) (int64, error)
 	DeleteAIRunsBefore(ctx context.Context, cutoff string) (int64, error)
-	// Deletes the row only while it still holds refresh_token_ct, so a
-	// connection made after the caller read the row is kept. Returns the number
-	// of rows deleted.
-	DeleteIntegrationsGoogle(ctx context.Context, refreshTokenCt []byte) (int64, error)
+	DeleteIntegrationsGoogle(ctx context.Context) error
 	DeleteRecipeIngredients(ctx context.Context, recipeID string) error
 	DeleteRecipeSteps(ctx context.Context, recipeID string) error
 	FailJob(ctx context.Context, arg FailJobParams) (int64, error)

internal/db/settings.sql.go

@@ -10,20 +10,14 @@ import (
 	"database/sql"
 )
 
-const deleteIntegrationsGoogle = `-- name: DeleteIntegrationsGoogle :execrows
+const deleteIntegrationsGoogle = `-- name: DeleteIntegrationsGoogle :exec
 DELETE FROM integrations_google
-WHERE id = 1 AND refresh_token_ct = ?
+WHERE id = 1
 `
 
-// Deletes the row only while it still holds refresh_token_ct, so a
-// connection made after the caller read the row is kept. Returns the number
-// of rows deleted.
-func (q *Queries) DeleteIntegrationsGoogle(ctx context.Context, refreshTokenCt []byte) (int64, error) {
-	result, err := q.db.ExecContext(ctx, deleteIntegrationsGoogle, refreshTokenCt)
-	if err != nil {
-		return 0, err
-	}
-	return result.RowsAffected()
+func (q *Queries) DeleteIntegrationsGoogle(ctx context.Context) error {
+	_, err := q.db.ExecContext(ctx, deleteIntegrationsGoogle)
+	return err
 }
 
 const getAISettings = `-- name: GetAISettings :one

internal/gcal/connect.go

@@ -20,6 +20,7 @@ import (
 	"net/url"
 	"slices"
 	"strings"
+	"sync"
 	"time"
 
 	"golang.org/x/oauth2"
@@ -82,9 +83,6 @@ var (
 	ErrScopes = errors.New("gcal: Google did not grant every Calendar permission")
 	// ErrNotConnected means no Google account is connected.
 	ErrNotConnected = errors.New("gcal: Google Calendar is not connected")
-	// ErrChanged means the connection changed while it was being
-	// disconnected.
-	ErrChanged = errors.New("gcal: Google Calendar connection changed during disconnect")
 )
 
 // Config is what NewService needs to talk to Google.
@@ -106,6 +104,8 @@ type Config struct {
 }
 
 // Service connects and disconnects the household's Google account.
+// MealQuest runs as one process, so the service's in-memory lock and pending
+// states see every request.
 type Service struct {
 	st        *store.Store
 	box       *secretbox.Box
@@ -115,6 +115,19 @@ type Service struct {
 	cookie    auth.FlowCookie
 	// now is the clock; tests replace it.
 	now func() time.Time
+
+	// account serializes every change to the connected account — a
+	// connect's exchange and store, a disconnect's read, revoke and delete —
+	// across the request to Google and the database write. Google revokes the
+	// whole grant, so a revoke overlapping a connect would also kill the
+	// token the connect stores.
+	account sync.Mutex
+
+	// pending holds the state of every connect started and not yet
+	// completed, with when it expires. Complete consumes a state once, so a
+	// replayed cookie finds it gone.
+	pendingMu sync.Mutex
+	pending   map[string]time.Time
 }
 
 // NewService returns the service over st, sealing refresh tokens with box.
@@ -147,6 +160,7 @@ func NewService(st *store.Store, box *secretbox.Box, cfg Config) (*Service, erro
 		client:    &http.Client{Timeout: googleTimeout, CheckRedirect: auth.CheckRedirect},
 		cookie:    auth.NewFlowCookie(connectCookieName, connectCookieKeyLabel, connectCookieTTL, cfg.SecretKey),
 		now:       time.Now,
+		pending:   map[string]time.Time{},
 	}, nil
 }
 
@@ -180,9 +194,11 @@ func (s *Service) Connect(w http.ResponseWriter, session auth.Session) (string,
 		Verifier: oauth2.GenerateVerifier(),
 		Session:  sessionBinding(session),
 	}
-	if err := s.cookie.Set(w, &state, s.now()); err != nil {
+	now := s.now()
+	if err := s.cookie.Set(w, &state, now); err != nil {
 		return "", fmt.Errorf("gcal: set connect cookie: %w", err)
 	}
+	s.addPending(state.State, now)
 	return s.oauth.AuthCodeURL(state.State,
 		oauth2.AccessTypeOffline,
 		oauth2.ApprovalForce,
@@ -191,12 +207,12 @@ func (s *Service) Connect(w http.ResponseWriter, session auth.Session) (string,
 }
 
 // Complete finishes the connect Google redirected r back with. It clears the
-// connect cookie, checks the state against it and session, exchanges the
-// code and stores the refresh token, sealed, with connected_at now. A
-// reconnect replaces the stored token and keeps the chosen calendar.
+// connect cookie, checks the state against it and session, consumes the
+// state, exchanges the code and stores the refresh token, sealed, with
+// connected_at now. A reconnect replaces the stored token and keeps the
+// chosen calendar.
 func (s *Service) Complete(w http.ResponseWriter, r *http.Request, session auth.Session) error {
 	ctx := r.Context()
-	// The connect cookie is single-use.
 	s.cookie.Clear(w)
 
 	var state connectState
@@ -210,6 +226,11 @@ func (s *Service) Complete(w http.ResponseWriter, r *http.Request, session auth.
 	if subtle.ConstantTimeCompare([]byte(sessionBinding(session)), []byte(state.Session)) != 1 {
 		return fmt.Errorf("%w: started by another session", ErrBadState)
 	}
+	// A copy of the cookie verifies until it expires, so the state is
+	// single-use only because it is consumed here.
+	if !s.consumePending(state.State) {
+		return fmt.Errorf("%w: already used", ErrBadState)
+	}
 	if q.Get("error") != "" {
 		// error and error_description are Google-supplied text; neither is
 		// kept.
@@ -220,6 +241,8 @@ func (s *Service) Complete(w http.ResponseWriter, r *http.Request, session auth.
 		return ErrNoCode
 	}
 
+	s.account.Lock()
+	defer s.account.Unlock()
 	token, err := s.oauth.Exchange(context.WithValue(ctx, oauth2.HTTPClient, s.client), code,
 		oauth2.VerifierOption(state.Verifier))
 	if err != nil {
@@ -238,6 +261,29 @@ func (s *Service) Complete(w http.ResponseWriter, r *http.Request, session auth.
 	return s.store(ctx, sealed)
 }
 
+// addPending records state, started at now, as pending until the connect
+// cookie expires, and forgets expired ones.
+func (s *Service) addPending(state string, now time.Time) {
+	s.pendingMu.Lock()
+	defer s.pendingMu.Unlock()
+	for k, expires := range s.pending {
+		if !now.Before(expires) {
+			delete(s.pending, k)
+		}
+	}
+	s.pending[state] = now.Add(connectCookieTTL)
+}
+
+// consumePending removes state from the pending states, reporting whether
+// it was there.
+func (s *Service) consumePending(state string) bool {
+	s.pendingMu.Lock()
+	defer s.pendingMu.Unlock()
+	_, ok := s.pending[state]
+	delete(s.pending, state)
+	return ok
+}
+
 // checkScopes fails with ErrScopes unless token's granted scopes include
 // every scope in Scopes. Google lets the household untick a permission on
 // the consent screen and reports what was granted in "scope".
@@ -286,6 +332,8 @@ func (s *Service) store(ctx context.Context, sealed []byte) error {
 // MealQuest's access in their Google account) counts as revoked. Any other
 // revoke failure keeps the connection so the household can retry.
 func (s *Service) Disconnect(ctx context.Context) error {
+	s.account.Lock()
+	defer s.account.Unlock()
 	row, err := db.New(s.st.Reader).GetIntegrationsGoogle(ctx)
 	if errors.Is(err, sql.ErrNoRows) {
 		return ErrNotConnected
@@ -300,13 +348,9 @@ func (s *Service) Disconnect(ctx context.Context) error {
 	if err := s.revoke(ctx, string(token)); err != nil {
 		return err
 	}
-	n, err := db.New(s.st.Writer).DeleteIntegrationsGoogle(ctx, row.RefreshTokenCt)
-	if err != nil {
+	if err := db.New(s.st.Writer).DeleteIntegrationsGoogle(ctx); err != nil {
 		return fmt.Errorf("gcal: delete Google integration: %w", err)
 	}
-	if n == 0 {
-		return ErrChanged
-	}
 	return nil
 }
 

internal/gcal/connect_test.go

@@ -290,6 +290,31 @@ func TestCompleteRefusesBadState(t *testing.T) {
 	}
 }
 
+// TestCompleteRefusesReplayedState presents a connect's cookie and state a
+// second time, with a second valid code from the same consent URL, as an
+// attacker holding a copy of the cookie could. The server refuses it,
+// whatever the browser did with the cleared cookie.
+func TestCompleteRefusesReplayedState(t *testing.T) {
+	t.Parallel()
+	h := newHarness(t)
+	c := h.connect(t, session)
+	if _, err := h.callback(c, session, h.consented(c)); err != nil {
+		t.Fatalf("Complete: %v", err)
+	}
+	stored, _ := h.row(t)
+
+	_, err := h.callback(c, session, h.consented(c))
+	if !errors.Is(err, gcal.ErrBadState) {
+		t.Fatalf("replayed Complete error = %v, want ErrBadState", err)
+	}
+	if got := h.google.TokenCalls(); got != 1 {
+		t.Errorf("token endpoint called %d times, want once: the replayed code was exchanged", got)
+	}
+	if row, _ := h.row(t); !bytes.Equal(row.RefreshTokenCt, stored.RefreshTokenCt) {
+		t.Error("the replay replaced the stored token")
+	}
+}
+
 func TestCompleteRefusesMissingRefreshToken(t *testing.T) {
 	t.Parallel()
 	h := newHarness(t)
@@ -371,6 +396,91 @@ func TestDisconnectRevokesAndClears(t *testing.T) {
 	}
 }
 
+// TestDisconnectRacingReconnect drives a disconnect and a reconnect into
+// each other at Google, with distinct tokens. However they interleave, the
+// end state is consistent: disconnected, or connected with a token Google
+// has not revoked.
+func TestDisconnectRacingReconnect(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name string
+		// hold stops the request the other operation must not slip past.
+		hold func(*gcaltest.Google) (<-chan struct{}, func())
+		// first is held at Google while second runs.
+		first, second func(h *harness, c connect, q url.Values) error
+	}{
+		{
+			// The disconnect has read the stored token and is revoking it
+			// when the reconnect would store a new one.
+			name:   "reconnect during revoke",
+			hold:   (*gcaltest.Google).HoldRevoke,
+			first:  disconnect,
+			second: reconnect,
+		},
+		{
+			// The reconnect's new token is issued but not yet stored when
+			// the disconnect would revoke the old one.
+			name:   "disconnect during exchange",
+			hold:   (*gcaltest.Google).HoldToken,
+			first:  reconnect,
+			second: disconnect,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			h := newHarness(t)
+			first := h.connect(t, session)
+			if _, err := h.callback(first, session, h.consented(first)); err != nil {
+				t.Fatal(err)
+			}
+			c := h.connect(t, session)
+			q := h.consented(c)
+
+			arrived, release := tc.hold(h.google)
+			defer release()
+			firstDone := make(chan error, 1)
+			go func() { firstDone <- tc.first(h, c, q) }()
+			<-arrived
+			secondDone := make(chan error, 1)
+			go func() { secondDone <- tc.second(h, c, q) }()
+			// Let the second operation run as far as it can while the first
+			// is held at Google: to the end when nothing serializes them.
+			select {
+			case err := <-secondDone:
+				secondDone <- err
+			case <-time.After(500 * time.Millisecond):
+			}
+			release()
+			for _, done := range []chan error{firstDone, secondDone} {
+				if err := <-done; err != nil {
+					t.Logf("operation: %v", err)
+				}
+			}
+
+			row, ok := h.row(t)
+			if !ok {
+				return
+			}
+			plain, err := h.box.Decrypt(row.RefreshTokenCt)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !h.google.Live(string(plain)) {
+				t.Errorf("stored as connected with a revoked token (Google revoked %d tokens)", len(h.google.Revoked()))
+			}
+		})
+	}
+}
+
+func disconnect(h *harness, _ connect, _ url.Values) error {
+	return h.svc.Disconnect(context.Background())
+}
+
+func reconnect(h *harness, c connect, q url.Values) error {
+	_, err := h.callback(c, session, q)
+	return err
+}
+
 func TestDisconnectTokenAlreadyRevokedAtGoogle(t *testing.T) {
 	t.Parallel()
 	h := newHarness(t)

internal/gcal/gcaltest/gcaltest.go

@@ -1,7 +1,9 @@
 // Package gcaltest provides a fake Google OAuth server for tests of the
 // Google Calendar connection: a token endpoint that checks the client, the
-// redirect URI and PKCE, and a revoke endpoint that records what it revoked.
-// It never touches the network beyond its own httptest server.
+// redirect URI and PKCE and issues a distinct refresh token per exchange, and
+// a revoke endpoint that, like Google's, revokes the whole grant: every token
+// issued so far. It never touches the network beyond its own httptest
+// server.
 package gcaltest
 
 import (
@@ -10,6 +12,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -19,7 +22,9 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/gcal"
 )
 
-// Fixed credentials and tokens the fake issues and expects.
+// Fixed credentials and tokens the fake issues and expects. The first
+// exchange issues RefreshToken; exchange n after it issues RefreshToken
+// followed by "-n", so every issued refresh token contains RefreshToken.
 const (
 	ClientID     = "mealquest-test.apps.googleusercontent.com"
 	ClientSecret = "google-client-s3cret-4f1c"
@@ -49,6 +54,19 @@ type Google struct {
 	codes        map[string]string // code -> PKCE challenge
 	tokenCalls   int
 	revoked      []string
+	// issued is every refresh token issued, in order; dead is those a
+	// revoke has invalidated.
+	issued []string
+	dead   map[string]bool
+	// tokenHold and revokeHold, when set, stop the next request to that
+	// endpoint.
+	tokenHold, revokeHold *hold
+}
+
+// hold stops one request: arrived is closed when it arrives, and it goes on
+// once release is closed.
+type hold struct {
+	arrived, release chan struct{}
 }
 
 // New starts a fake Google that requires redirectURL at its token endpoint.
@@ -61,6 +79,7 @@ func New(t *testing.T, redirectURL string) *Google {
 		tokenStatus:  http.StatusOK,
 		revokeStatus: http.StatusOK,
 		codes:        map[string]string{},
+		dead:         map[string]bool{},
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("POST /token", g.token)
@@ -138,6 +157,55 @@ func (g *Google) Revoked() []string {
 	return append([]string(nil), g.revoked...)
 }
 
+// Live reports whether token was issued by the fake and no revoke has
+// invalidated it since.
+func (g *Google) Live(token string) bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return slices.Contains(g.issued, token) && !g.dead[token]
+}
+
+// HoldToken stops the next token request after its refresh token is issued
+// and before it is answered, so a revoke can land in between. It returns a
+// channel closed when the request arrives and the function that lets it go
+// on.
+func (g *Google) HoldToken() (arrived <-chan struct{}, release func()) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.tokenHold = newHold()
+	return g.tokenHold.arrived, g.tokenHold.releaseFunc()
+}
+
+// HoldRevoke stops the next revoke request before Google acts on it. It
+// returns a channel closed when the request arrives and the function that
+// lets it go on.
+func (g *Google) HoldRevoke() (arrived <-chan struct{}, release func()) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.revokeHold = newHold()
+	return g.revokeHold.arrived, g.revokeHold.releaseFunc()
+}
+
+func newHold() *hold {
+	return &hold{arrived: make(chan struct{}), release: make(chan struct{})}
+}
+
+func (h *hold) releaseFunc() func() {
+	var once sync.Once
+	return func() { once.Do(func() { close(h.release) }) }
+}
+
+// wait stops the request at h, if any, without holding g.mu.
+func (g *Google) wait(h *hold) {
+	if h == nil {
+		return
+	}
+	g.mu.Unlock()
+	close(h.arrived)
+	<-h.release
+	g.mu.Lock()
+}
+
 func (g *Google) token(w http.ResponseWriter, r *http.Request) {
 	g.mu.Lock()
 	defer g.mu.Unlock()
@@ -168,8 +236,16 @@ func (g *Google) token(w http.ResponseWriter, r *http.Request) {
 	}
 	refresh := ""
 	if !g.omitRefresh {
-		refresh = fmt.Sprintf(`"refresh_token":%q,`, RefreshToken)
+		issue := RefreshToken
+		if n := len(g.issued); n > 0 {
+			issue = fmt.Sprintf("%s-%d", RefreshToken, n+1)
+		}
+		g.issued = append(g.issued, issue)
+		refresh = fmt.Sprintf(`"refresh_token":%q,`, issue)
 	}
+	h := g.tokenHold
+	g.tokenHold = nil
+	g.wait(h)
 	w.Header().Set("Content-Type", "application/json")
 	fmt.Fprintf(w, `{"access_token":%q,%s"expires_in":3599,"scope":%q,"token_type":"Bearer"}`,
 		AccessToken, refresh, g.scope)
@@ -186,8 +262,18 @@ func (g *Google) revoke(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "bad form", http.StatusBadRequest)
 		return
 	}
+	h := g.revokeHold
+	g.revokeHold = nil
+	g.wait(h)
 	token := r.PostForm.Get("token")
 	g.revoked = append(g.revoked, token)
+	if g.revokeStatus == http.StatusOK && slices.Contains(g.issued, token) {
+		// Google revokes the grant, not just the token: every token issued
+		// to it so far stops working.
+		for _, t := range g.issued {
+			g.dead[t] = true
+		}
+	}
 	body := g.revokeBody
 	if body == "" && g.revokeStatus != http.StatusOK {
 		body = "cannot revoke " + token

internal/server/google.go

@@ -90,9 +90,6 @@ func deleteGoogle(logger *slog.Logger, svc *gcal.Service) http.HandlerFunc {
 			w.WriteHeader(http.StatusNoContent)
 		case errors.Is(err, gcal.ErrNotConnected):
 			writeError(w, r, logger, http.StatusConflict, "not_connected", "Google Calendar is not connected")
-		case errors.Is(err, gcal.ErrChanged):
-			writeError(w, r, logger, http.StatusConflict, "conflict",
-				"the Google Calendar connection changed; reload and try again")
 		case errors.As(err, &revoke):
 			logger.WarnContext(r.Context(), "revoke Google access", "error", err)
 			writeError(w, r, logger, http.StatusBadGateway, "revoke_failed",

migrations/queries/settings.sql

@@ -106,12 +106,9 @@ ON CONFLICT (id) DO UPDATE SET
 	updated_at = excluded.updated_at
 RETURNING *;
 
--- name: DeleteIntegrationsGoogle :execrows
--- Deletes the row only while it still holds refresh_token_ct, so a
--- connection made after the caller read the row is kept. Returns the number
--- of rows deleted.
+-- name: DeleteIntegrationsGoogle :exec
 DELETE FROM integrations_google
-WHERE id = 1 AND refresh_token_ct = ?;
+WHERE id = 1;
 
 -- name: GetChatSettings :one
 -- Returns sql.ErrNoRows until the row exists.