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>

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)