cli: address Bugbot/Copilot review on the cell fan-out layer

9ef8a5c·

Soph·1w ago·4 files·+169 added/-29 removed

• Bound resolveCellBaseURLs's catalog lookup with cellResolveTimeout, like resolveRepoCellTarget: a hung core must not stall the command before the fan-out even starts.
• Refuse a concrete baseURL without a jurisdiction: minting a home-jurisdiction token and dialing a foreign cell with it is exactly the mismatch resolveRepoCellTarget refuses; such a group stays on home routing.
• Key groupReposByCell on (cell, jurisdiction): blank-cell index rows in different jurisdictions no longer collapse into one group routed by whichever repo came first.
• Single-flight token mints per jurisdiction (mintSlot) instead of one factory-wide mutex held across the exchange: a slow region's mint no longer burns other cells' per-cell deadlines. Failed mints cache nothing, so the next caller retries.

Co-Authored-By: Claude Fable 5 noreply@anthropic.com

Sessions

ab23b05bf985View transcript

Changes

4

• cmd/entire/cli

• auth

  • Mcell_data_api.go+29/-11

• Mcell_data_api_test.go+68

• Mcell_fanout.go+37/-11

• Mcell_fanout_test.go+35/-7

118 unmodified lines

119
120
121
122
123
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
4 unmodified lines

141
142
143
134
144
145
146
147
31 unmodified lines

179
180
181
172
173
174
175
182
183
184
185
186
187
178
179
180
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
187
205
206
207
208

118 unmodified lines

type CellClientFactory struct {
    subject cellSubject

mu     sync.Mutex
    tokens map[string]string // jurisdiction -> minted identity token
    mu    sync.Mutex           // guards slots (map access only, never held across I/O)
    slots map[string]*mintSlot // jurisdiction -> its token slot
}

// mintSlot single-flights one jurisdiction's token: the slot mutex is held
// across the mint, so concurrent callers for the same jurisdiction wait for one
// exchange instead of duplicating it, while other jurisdictions mint in
// parallel on their own slots. A failed mint caches nothing — the next caller
// retries.
type mintSlot struct {
    mu    sync.Mutex
    token string
}

// NewEntireAPICellClientFactory resolves the exchange subject (active stored
4 unmodified lines

if err != nil {
        return nil, err
    }
    return &CellClientFactory{subject: subject, tokens: make(map[string]string)}, nil
    return &CellClientFactory{subject: subject, slots: make(map[string]*mintSlot)}, nil
}

// ClientFor returns an authenticated client for the given cell target (nil
31 unmodified lines

// tokenFor returns the cached identity token for jurisdiction, minting it on
// first use. The mutex is held across the mint: concurrent callers for the
// same jurisdiction wait for one exchange instead of duplicating it, at the
// cost of serializing cross-jurisdiction mints (fine for the handful of
// jurisdictions a fan-out touches).
// first use. Locking is per jurisdiction (see mintSlot): a slow exchange for
// one jurisdiction never blocks another jurisdiction's mint — in a fan-out,
// healthy cells must not burn their per-cell deadline waiting on an unrelated
// region's exchange.
func (f *CellClientFactory) tokenFor(ctx context.Context, jurisdiction, coreURL string) (string, error) {
    f.mu.Lock()
    defer f.mu.Unlock()
    if token, ok := f.tokens[jurisdiction]; ok {
        return token, nil
    }
    slot, ok := f.slots[jurisdiction]
    if !ok {
        slot = &mintSlot{}
        f.slots[jurisdiction] = slot
    }
    f.mu.Unlock()

slot.mu.Lock()
    defer slot.mu.Unlock()
    if slot.token != "" {
        return slot.token, nil
    }
    audience := jurisdictionAudience(jurisdiction, f.subject.dataOrigin, f.subject.discoveredCore)
    token, err := exchangeJurisdictionToken(ctx, coreURL, f.subject.loginJWT, audience, f.subject.httpClient)
    if err != nil {
        return "", fmt.Errorf("exchange jurisdictional identity token: %w", err)
    }
    f.tokens[jurisdiction] = token
    slot.token = token
    return token, nil
}

Mcmd/entire/cli/auth/cell_data_api.go+29/-11

10 unmodified lines

11
12
13
14
15
16
17
587 unmodified lines

605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674

10 unmodified lines

"net/url"
    "path/filepath"
    "strings"
    "sync"
    "testing"
    "time"

587 unmodified lines

t.Fatalf("exchange audiences = %q, want %q", got, want)
    }
}

// TestCellClientFactory_SingleFlightsConcurrentMints pins the per-jurisdiction
// locking: concurrent ClientFor calls for the same jurisdiction produce exactly
// one exchange (the rest wait on the slot and reuse the result), not one each.
// Not parallel: manipulates env + token store.
func TestCellClientFactory_SingleFlightsConcurrentMints(t *testing.T) {
    t.Setenv("ENTIRE_CONFIG_DIR", t.TempDir())
    t.Setenv("ENTIRE_API_BASE_URL", "https://entire.io")
    t.Setenv("ENTIRE_API_AUDIENCE_TEMPLATE", "")
    t.Setenv("ENTIRE_CORE_BASE_URL_TEMPLATE", "")
    restore := tokenstore.UseFileBackendForTesting(filepath.Join(t.TempDir(), "tokens.json"))
    t.Cleanup(restore)

var mu sync.Mutex
    exchangeCount := 0
    coreSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.URL.Path != oauthTokenPath {
            http.NotFound(w, r)
            return
        }
        mu.Lock()
        exchangeCount++
        mu.Unlock()
        time.Sleep(30 * time.Millisecond) // widen the window concurrent callers could race into
        w.Header().Set("Content-Type", "application/json")
        _, _ = fmt.Fprint(w, `{"access_token":"identity-token","token_type":"Bearer","expires_in":3600}`)
    }))
    defer coreSrv.Close()

svc := tokenstore.CoreKeyringService(coreSrv.URL)
    loginJWT := makeJWT(t, fmt.Sprintf(`{"iss":%q,"home_jurisdiction":"us","exp":%d}`, coreSrv.URL, time.Now().Add(2*time.Hour).Unix()))
    if err := tokenstore.Set(svc, "me", tokenstore.EncodeTokenWithExpiration(loginJWT, 7200)); err != nil {
        t.Fatalf("seed token: %v", err)
    }
    ctxObj := &contexts.Context{Name: "me@core", CoreURL: coreSrv.URL, Handle: "me", KeychainService: svc}
    t.Cleanup(SetResolveContextForCellAPIForTest(t, func(context.Context, string, string, string, *http.Client, clusterdiscovery.DebugFunc) (*contexts.Context, error) {
        return ctxObj, nil
    }))
    t.Cleanup(SetCellExchangeTransportForTest(t, coreSrv.Client().Transport))

factory, err := NewEntireAPICellClientFactory(context.Background(), false)
    if err != nil {
        t.Fatalf("NewEntireAPICellClientFactory: %v", err)
    }

var wg sync.WaitGroup
    err := make([]error, 4)
    for i := range errs {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            _, errs[i] = factory.ClientFor(context.Background(), &CellTarget{
                BaseURL:      fmt.Sprintf("https://cell-%d.api.example", i),
                Jurisdiction: "eu",
            })
        }(i)
    }
    wg.Wait()
    for i, err := range errs {
        if err != nil {
            t.Fatalf("ClientFor[%d]: %v", i, err)
        }
    }
    if exchangeCount != 1 {
        t.Fatalf("exchange count = %d, want 1 (same jurisdiction single-flighted)", exchangeCount)
    }
}

Mcmd/entire/cli/auth/cell_data_api_test.go+68

46 unmodified lines

47
48
49
50
51
50
51
52
53
54
55
56
57
1 unmodified line

59
60
61
59
62
63
64
65
66
67
63
68
69
65
70
71
72
73
3 unmodified lines

77
78
79
75
80
81
82
83
84
85
86
87
88
89
80
81
82
90
91
92
93
94
95
96
97
98
99
100
101
10 unmodified lines

112
113
114
100
115
116
117
118
119
120
102
121
122
123
124
125
126
127
128
129
130
131
132

46 unmodified lines

}

// groupReposByCell groups a repo index by hosting cell, one group per distinct
// cell, deterministically ordered by cell name. Entries without an ID are
// skipped (nothing to ask the cell about).
// cell, deterministically ordered by cell name (jurisdiction as tiebreak).
// Entries without an ID are skipped (nothing to ask the cell about). The key
// includes the jurisdiction so entries whose index row carries no cell don't
// collapse across jurisdictions into one group routed by whichever repo came
// first — they stay per-jurisdiction and route via the jurisdiction fallback.
func groupReposByCell(repos []coreapi.RepoIndexEntry) []cellGroup {
    byCell := make(map[string]*cellGroup)
    for _, r := range repos {
1 unmodified line

if id == "" {
            continue
        }
        key := strings.ToLower(strings.TrimSpace(r.Cell))
        cell := strings.ToLower(strings.TrimSpace(r.Cell))
        jurisdiction := strings.ToLower(strings.TrimSpace(r.Jurisdiction))
        key := cell + "\x00" + jurisdiction
        g, ok := byCell[key]
        if !ok {
            g = &cellGroup{
                cell:         key,
                cell:         cell,
                clusterSlug:  strings.ToLower(strings.TrimSpace(r.ClusterSlug)),
                jurisdiction: strings.ToLower(strings.TrimSpace(r.Jurisdiction)),
                jurisdiction: jurisdiction,
            }
            byCell[key] = g
        }
3 unmodified lines

for _, g := range byCell {
        cells = append(cells, *g)
    }
    sort.Slice(cells, func(i, j int) bool { return cells[i].cell < cells[j].cell })
sort.Slice(cells, func(i, j int) bool {
    if cells[i].cell != cells[j].cell {
        return cells[i].cell < cells[j].cell
    }
    return cells[i].jurisdiction < cells[j].jurisdiction
})
    return cells
}

// resolveCellBaseURLs fills each group's baseURL from the cluster catalog,
// joining on ClusterSlug ↔ Cluster.Slug. Best-effort: on a catalog error or a
// missing/incomplete cluster row the group keeps baseURL "" and falls back to
// jurisdiction routing — a degraded catalog must not sink the fan-out.
// joining on ClusterSlug ↔ Cluster.Slug. Best-effort: on a catalog error or
// timeout (bounded by cellResolveTimeout, like resolveRepoCellTarget — a hung
// core must not stall the command) or a missing/incomplete cluster row, the
// group keeps baseURL "" and falls back to jurisdiction routing — a degraded
// catalog must not sink the fan-out.
func resolveCellBaseURLs(ctx context.Context, c cellCoreClient, cells []cellGroup) {
    ctx, cancel := context.WithTimeout(ctx, cellResolveTimeout)
    defer cancel()

clusters, err := c.ListClusters(ctx)
    if err != nil {
        logging.Debug(ctx, "cell fan-out: list clusters failed, using jurisdiction routing", "error", err.Error())
10 unmodified lines

"cluster_slug", cells[i].clusterSlug, "cell", cells[i].cell)
            continue
        }
        cells[i].baseURL = strings.TrimRight(strings.TrimSpace(cl.ApiUrl.Or("").String()), "/")
        // A concrete baseURL needs a jurisdiction to mint the matching token
        // for — mirroring resolveRepoCellTarget, which refuses a target unless
        // both are present. Setting baseURL with an unknown jurisdiction would
        // dial the cell with a home-jurisdiction token.
        jurisdiction := cells[i].jurisdiction
        if j := strings.ToLower(strings.TrimSpace(cl.Jurisdiction)); j != "" {
            cells[i].jurisdiction = j
            jurisdiction = j
        }
        if jurisdiction == "" {
            logging.Debug(ctx, "cell fan-out: no jurisdiction for cluster, using home routing",
                "cluster_slug", cells[i].clusterSlug, "cell", cells[i].cell)
            continue
        }
        cells[i].jurisdiction = jurisdiction
        cells[i].baseURL = strings.TrimRight(strings.TrimSpace(cl.ApiUrl.Or("").String()), "/")
    }
}