From 77510766de7ba496c3a0198e1c9d1862f93c680d Mon Sep 17 00:00:00 2001 From: Arjun Mehta Date: Thu, 27 Aug 2026 14:25:13 +0000 Subject: [PATCH 1/3] perf(controller): eliminate redundant treehash storage reads GetChangedTargets previously read the same treehash values multiple times during cache lookup and background cache population. Read the treehashes once and pass the values through the existing call chain so they can be reused. This reduces treehash storage.Get calls from 6 to 2 in the affected cache miss path. --- controller/getchangedtargets.go | 58 ++++++----- controller/getchangedtargets_test.go | 138 +++++++++++++++++++++++++-- controller/gettargetgraph.go | 50 ++++++---- 3 files changed, 192 insertions(+), 54 deletions(-) diff --git a/controller/getchangedtargets.go b/controller/getchangedtargets.go index cefe698e..94cbf7a9 100644 --- a/controller/getchangedtargets.go +++ b/controller/getchangedtargets.go @@ -76,6 +76,7 @@ type job struct { completed bool ctx context.Context cancel context.CancelCauseFunc + treehash string } // GetChangedTargets returns the changed targets between two revisions. If the @@ -113,9 +114,16 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str maxDist = request.GetOutputConfig().GetMaxDistance() } + // Read treehashes once for the entire request. These are passed through + // the request pipeline to avoid redundant storage reads. + treehash1, treehash2, err := readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), e, opGetChangedTargets) + if err != nil { + return fmt.Errorf("read revision treehash: %w", err) + } + // Fast path: stream a previously computed result straight from cache. if !request.GetBypassCache() { - served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start) + served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start, treehash1, treehash2) if err != nil { return fmt.Errorf("serve from cache: %w", err) } @@ -125,7 +133,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str } // Fetch both revisions' target graphs concurrently. - firstGraph, secondGraph, err := c.fetchTargetGraphs(ctx, e, logger, request) + firstGraph, secondGraph, err := c.fetchTargetGraphs(ctx, e, logger, request, treehash1, treehash2) if err != nil { return fmt.Errorf("fetch target graphs: %w", err) } @@ -142,7 +150,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str } // Cache the computed result concurrently so it doesn't block the stream send. - c.cacheComparedTargets(logger, request, changedTargetsResponses) + c.cacheComparedTargets(logger, request, changedTargetsResponses, treehash1, treehash2) sendStart := time.Now() if err := sendTrimmedChangedTargets(stream, changedTargetsResponses, maxDist, request.GetOutputConfig()); err != nil { @@ -164,15 +172,18 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // - (false, nil) on a cache miss or a corrupt blob — the caller should recompute; // - (false, err) on an infra failure or a client disconnect that aborts the request. // -// readTreehash returns ("", nil) on a cache miss (skip cache, recompute) but any -// real storage error surfaces here so an infra failure that disables the cache -// (e.g. a missing-deadline "missing TTL" reject) becomes a visible request failure -// rather than silent degradation. -func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time) (bool, error) { +// If treehash1 and treehash2 are provided (non-empty), they are used instead of +// reading from storage. This allows the caller to reuse treehash values already +// read earlier in the request. +func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time, treehash1, treehash2 string) (bool, error) { cacheStart := time.Now() - treehash1, treehash2, err := readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), e, opGetChangedTargets) - if err != nil { - return false, fmt.Errorf("read revision treehash: %w", err) + if treehash1 == "" || treehash2 == "" { + // Treehashes not provided by caller; read them now. + var err error + treehash1, treehash2, err = readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), e, opGetChangedTargets) + if err != nil { + return false, fmt.Errorf("read revision treehash: %w", err) + } } if treehash1 == "" || treehash2 == "" { return false, nil @@ -241,13 +252,17 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metric // original failure is returned. A client disconnect surfaces as a user-cancelled // error. A graph stored as a TGB blob comes back as its undrained reader; a // gob-era graph is drained into chunks here, inside the concurrent fetch. -func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest) (fetchedGraph, fetchedGraph, error) { +func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, treehash1, treehash2 string) (fetchedGraph, fetchedGraph, error) { jobs := make([]*job, 2) for i := 0; i < 2; i++ { // create independent contexts for each job; if one of the jobs fails, the other one should be cancelled to save resources and improve latency ctxNew, cancelNew := context.WithCancelCause(ctx) defer cancelNew(nil) - jobs[i] = &job{ctx: ctxNew, cancel: cancelNew} + th := treehash1 + if i == 1 { + th = treehash2 + } + jobs[i] = &job{ctx: ctxNew, cancel: cancelNew, treehash: th} } // Start jobs for both revisions. Success or failure, the result will report to the results channel. @@ -284,7 +299,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, ExcludeFilesRegex: request.GetRequestOptions().GetExtraExcludeFilesRegex(), BypassCache: request.GetBypassCache(), } - graphReader, err := c.getGraph(jobs[idx].ctx, e, entityReq) + graphReader, err := c.getGraph(jobs[idx].ctx, e, entityReq, jobs[idx].treehash) if err != nil || graphReader == nil { results <- graphResult{order: idx, err: err} return @@ -376,7 +391,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, // a fire-and-forget goroutine so it does not block the stream send. The responses // is only read (never mutated) by the goroutine and the foreground send, so // concurrent access is safe; the caller must not mutate it. This is best effort. -func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetChangedTargetsRequest, responses []entity.GetChangedTargetsResponse) { +func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetChangedTargetsRequest, responses []entity.GetChangedTargetsResponse, treehash1, treehash2 string) { go func() { // Use c.appCtx directly: the cache write is fire-and-forget and must // outlive the request (so a client disconnect doesn't abort it) but @@ -385,17 +400,8 @@ func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetCha // is cancelled on shutdown. Per-operation deadlines are the storage // backend's responsibility — the controller is backend-agnostic and // must not encode any one implementation's I/O budget. - // The treehash reads here are for building the write key, not a cache - // serve attempt, so they pass a no-op emitter to avoid skewing the - // treehash cache hit rate. - treehash1, treehash2, err := readTreehashParallel(c.appCtx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), metrics.Nop(), opGetChangedTargets) - if err != nil { - // Goroutine outlives the handler so we can't return; log loudly and - // abandon the cache write. Surfacing infra failures matters more than - // a missed cache opportunity. - logger.Warn("GetChangedTargets: skipping cache write, failed to read revision treehash", zap.Error(err)) - return - } + // The treehash values are passed by the caller (already read during + // the request) to avoid redundant storage reads. if treehash1 != "" && treehash2 != "" { cacheKey := cachekey.GetComparedTargetsCachePath(request.GetFirstRevision().GetRemote(), treehash1, treehash2, request.GetRequestOptions().GetExtraExcludeFilesRegex()) if writeErr := storage.WriteChangedTargetsStream(c.appCtx, c.storage, cacheKey, responses); writeErr != nil { diff --git a/controller/getchangedtargets_test.go b/controller/getchangedtargets_test.go index 19c74108..dbaacd76 100644 --- a/controller/getchangedtargets_test.go +++ b/controller/getchangedtargets_test.go @@ -538,8 +538,8 @@ func TestGetChangedTargets_streamChunks(t *testing.T) { default: return storage.DownloadResponse{}, fmt.Errorf("unexpected key: %s", req.Key) } - // readTreehash (×2 pre) + comparison cache miss (×1) + graph computation (×4) + readTreehash (×2 post) = 9 - }).Times(9) + // readTreehash (×2 pre) + comparison cache miss (×1) + graph computation (×2) = 5 + }).Times(5) // Put is launched in a goroutine — use a channel to wait for it before the test ends. putDone := make(chan struct{}, 1) storagemock.EXPECT().Put(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.UploadRequest) error { @@ -1408,7 +1408,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { c.storage = st stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now()) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), "", "") require.NoError(t, err) assert.False(t, served, "a cache miss must not be served") }) @@ -1447,7 +1447,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) // No Send expectation: a corrupt blob must not send anything to the client. - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now()) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), "", "") require.NoError(t, err) assert.False(t, served, "a corrupt blob must trigger recompute, not a partial send") }) @@ -1481,7 +1481,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now()) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), "", "") require.NoError(t, err) assert.True(t, served, "a clean cache hit must be served") }) @@ -1508,7 +1508,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) + first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") require.NoError(t, err) require.Len(t, first.chunks, 1) require.Len(t, second.chunks, 1) @@ -1532,7 +1532,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) + first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") require.Error(t, err) assert.ErrorIs(t, err, injected) assert.Zero(t, first) @@ -1553,7 +1553,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) + _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") require.Error(t, err) }) @@ -1568,7 +1568,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) + _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") require.Error(t, err) }) } @@ -1685,3 +1685,123 @@ func TestSeedAttributesFor(t *testing.T) { assert.Equal(t, map[string]bool{"size": true, "timeout": true}, c.seedAttributesFor("some-remote")) }) } + +// TestGetChangedTargets_TreehashReadCount verifies that treehashes are read +// exactly once per revision during a GetChangedTargets request, eliminating +// redundant storage reads that previously occurred in three separate code paths: +// 1. serveChangedTargetsFromCache (compared-targets cache lookup) +// 2. getGraph (individual graph cache lookup) +// 3. cacheComparedTargets (background cache write) +func TestGetChangedTargets_TreehashReadCount(t *testing.T) { + ctrl := gomock.NewController(t) + stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) + stream.EXPECT().Context().Return(t.Context()) + + var sentResponses []*pb.GetChangedTargetsResponse + stream.EXPECT().Send(gomock.Any()).DoAndReturn(func(resp *pb.GetChangedTargetsResponse, opts ...interface{}) error { + sentResponses = append(sentResponses, resp) + return nil + }).Times(2) + + storagemock := storagemock.NewMockStorage(ctrl) + + // Build first revision graph (2 chunks: Targets + Metadata) + var buf1 bytes.Buffer + enc1 := gob.NewEncoder(&buf1) + enc1.Encode(entity.GetTargetGraphResponse{ + Targets: []entity.OptimizedTarget{ + {ID: 1, Hash: "h1", RuleType: 100}, + {ID: 2, Hash: "h2-old", RuleType: 300}, + }, + }) + enc1.Encode(entity.GetTargetGraphResponse{ + Metadata: &entity.Metadata{ + TargetIDMapping: map[int32]string{1: "//app:target1", 2: "//app:target2"}, + RuleTypeMapping: map[int32]string{100: "go_library", 300: "source file"}, + }, + }) + graph1Bytes := buf1.Bytes() + + // Build second revision graph - target2 has different hash + var buf2 bytes.Buffer + enc2 := gob.NewEncoder(&buf2) + enc2.Encode(entity.GetTargetGraphResponse{ + Targets: []entity.OptimizedTarget{ + {ID: 1, Hash: "h1", RuleType: 100}, + {ID: 2, Hash: "h2-new", RuleType: 300}, // changed hash + }, + }) + enc2.Encode(entity.GetTargetGraphResponse{ + Metadata: &entity.Metadata{ + TargetIDMapping: map[int32]string{1: "//app:target1", 2: "//app:target2"}, + RuleTypeMapping: map[int32]string{100: "go_library", 300: "source file"}, + }, + }) + graph2Bytes := buf2.Bytes() + + // Expected Get calls after optimization: + // - 2 treehash reads (READ A in serveChangedTargetsFromCache) + // - 1 compared-targets cache miss + // - 2 graph reads (using pre-read treehashes, no additional treehash reads) + // Total: 5 storage.Get calls + treehashReadCount := 0 + graphReadCount := 0 + comparedTargetsReadCount := 0 + + storagemock.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, req storage.DownloadRequest) (storage.DownloadResponse, error) { + switch { + case strings.Contains(req.Key, "compared-targets"): + comparedTargetsReadCount++ + return storage.DownloadResponse{}, storage.NewNotFoundError(req.Key) + case strings.Contains(req.Key, "treehashes"): + treehashReadCount++ + if strings.Contains(req.Key, "sha1") { + return storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader([]byte("treehash1")))}, nil + } + return storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader([]byte("treehash2")))}, nil + case strings.Contains(req.Key, "treehash1"): + graphReadCount++ + return storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader(graph1Bytes))}, nil + case strings.Contains(req.Key, "treehash2"): + graphReadCount++ + return storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader(graph2Bytes))}, nil + default: + return storage.DownloadResponse{}, fmt.Errorf("unexpected key: %s", req.Key) + } + }).AnyTimes() + + // Put is launched in a goroutine — use a channel to wait for it before the test ends. + putDone := make(chan struct{}, 1) + storagemock.EXPECT().Put(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.UploadRequest) error { + putDone <- struct{}{} + return nil + }) + + c := NewController(context.Background(), Params{ + Logger: zaptest.NewLogger(t), + Storage: storagemock, + Orchestrator: orchestratormock.NewMockOrchestrator(ctrl), + }) + + request := &pb.GetChangedTargetsRequest{ + FirstRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"}, + SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"}, + OutputConfig: &pb.OutputConfig{MaxDistance: -1, IncludeHashes: true, IncludeTags: true, IncludeAttributes: true}, + } + + err := c.GetChangedTargets(request, stream) + require.NoError(t, err) + + select { + case <-putDone: + case <-time.After(time.Second): + assert.Fail(t, "cache write goroutine did not complete in time") + } + + // Before optimization: treehashReadCount == 6 (3 per revision) + // After optimization: treehashReadCount == 2 (1 per revision, shared) + assert.Equal(t, 2, treehashReadCount, "treehash should be read exactly once per revision") + assert.Equal(t, 2, graphReadCount, "graph should be read once per revision") + assert.Equal(t, 1, comparedTargetsReadCount, "compared-targets cache consulted once") +} diff --git a/controller/gettargetgraph.go b/controller/gettargetgraph.go index e59ce2ea..257ea6c4 100644 --- a/controller/gettargetgraph.go +++ b/controller/gettargetgraph.go @@ -56,7 +56,7 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb if err != nil { return tangoerrors.NewUser(fmt.Errorf("convert get target graph request: %w", err)) } - graphReader, err := c.getGraph(ctx, e, entityReq) + graphReader, err := c.getGraph(ctx, e, entityReq, "") if err != nil { return fmt.Errorf("get graph: %w", err) } @@ -96,34 +96,46 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb // entries store the full payload and stripping happens at send time, so // letting an orchestrator see it could poison the shared cache with // stripped graphs. -func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entity.GetTargetGraphRequest) (storage.GraphReader, error) { +// If treehash is provided (non-empty), it is used instead of reading from storage. +func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entity.GetTargetGraphRequest, treehash string) (storage.GraphReader, error) { start := time.Now() logger := c.logger.With( zap.Any("build_description", req.Build), ) if !req.BypassCache { - // Look up the the git treehash based on cache path - treehashCachePath := cachekey.GetTreehashCachePath(req.Build) - treehashResponse, err := c.storage.Get(ctx, storage.DownloadRequest{Key: treehashCachePath}) - metrics.RecordCacheLookup(e, opGetTargetGraph, metrics.TreehashCacheLookup, err) - if err != nil { - if storage.IsNotFound(err) { - // Cache miss - blob doesn't exist, need to compute and store target graph - logger.Debug("getGraph: treehash not found", zap.Error(err)) - } else { - // Other errors (network, infra issues) should be retried - return nil, fmt.Errorf("get treehash: %w", err) - } + var treehashValue string + if treehash != "" { + // Use the pre-read treehash value passed by the caller. + treehashValue = treehash + logger.Debug("getGraph: using pre-read treehash") } else { - defer func() { _ = treehashResponse.ReadCloser.Close() }() - treehashBytes, err := io.ReadAll(treehashResponse.ReadCloser) + // Look up the git treehash based on cache path + treehashCachePath := cachekey.GetTreehashCachePath(req.Build) + treehashResponse, err := c.storage.Get(ctx, storage.DownloadRequest{Key: treehashCachePath}) + metrics.RecordCacheLookup(e, opGetTargetGraph, metrics.TreehashCacheLookup, err) if err != nil { - return nil, fmt.Errorf("read treehash: %w", err) + if storage.IsNotFound(err) { + // Cache miss - blob doesn't exist, need to compute and store target graph + logger.Debug("getGraph: treehash not found", zap.Error(err)) + } else { + // Other errors (network, infra issues) should be retried + return nil, fmt.Errorf("get treehash: %w", err) + } + } else { + defer func() { _ = treehashResponse.ReadCloser.Close() }() + treehashBytes, err := io.ReadAll(treehashResponse.ReadCloser) + if err != nil { + return nil, fmt.Errorf("read treehash: %w", err) + } + treehashValue = string(treehashBytes) + logger.Info("getGraph: treehash found") } - logger.Info("getGraph: treehash found") + } + + if treehashValue != "" { // Download the target graph based on treehash. storageStart := time.Now() - graphReader, err := c.readCachedGraph(ctx, logger, req.Build.Remote, string(treehashBytes), req.Build.Strategy, req.ExcludeFilesRegex) + graphReader, err := c.readCachedGraph(ctx, logger, req.Build.Remote, treehashValue, req.Build.Strategy, req.ExcludeFilesRegex) if ctx.Err() != nil { err = context.Cause(ctx) } From 25ebe7530aa9b0e0eb041957b5ab3ae211d5f046 Mon Sep 17 00:00:00 2001 From: Arjun Mehta Date: Fri, 28 Aug 2026 03:55:16 +0000 Subject: [PATCH 2/3] refactor(controller): resolve treehashes once per request Replace the empty-string treehash sentinel parameters with a request-scoped resolver that reuses successfully resolved treehash values. This avoids repeating the same storage reads during GetChangedTargets while keeping treehash resolution and error handling request-local. --- controller/getchangedtargets.go | 123 +++++++++++++++++++-------- controller/getchangedtargets_test.go | 30 ++++--- controller/gettargetgraph.go | 20 +++-- 3 files changed, 117 insertions(+), 56 deletions(-) diff --git a/controller/getchangedtargets.go b/controller/getchangedtargets.go index 94cbf7a9..597ca904 100644 --- a/controller/getchangedtargets.go +++ b/controller/getchangedtargets.go @@ -20,6 +20,7 @@ import ( "fmt" "io" "maps" + "sync" "time" "github.com/uber/tango/core/cachekey" @@ -76,7 +77,56 @@ type job struct { completed bool ctx context.Context cancel context.CancelCauseFunc - treehash string + revision *pb.BuildDescription +} + +// treehashResolver memoizes treehash reads within a single request. +// It avoids redundant storage reads by caching successfully resolved +// treehash values. Errors are not cached — they are returned to the +// caller so that transient failures remain retryable. +type treehashResolver struct { + storage storage.Storage + emitter *metrics.Emitter + op string + mu sync.Mutex + cache map[string]string +} + +// newTreehashResolver creates a resolver for a single GetChangedTargets request. +func newTreehashResolver(st storage.Storage, e *metrics.Emitter, op string) *treehashResolver { + return &treehashResolver{ + storage: st, + emitter: e, + op: op, + cache: make(map[string]string), + } +} + +// resolve resolves the treehash for a build description. +// It memoizes successful reads; errors are not cached. +func (r *treehashResolver) resolve(ctx context.Context, build *pb.BuildDescription) (string, error) { + entityBuild, err := mapper.ProtoToBuildDescription(build) + if err != nil { + return "", err + } + key := cachekey.GetTreehashCachePath(entityBuild) + + r.mu.Lock() + if val, ok := r.cache[key]; ok { + r.mu.Unlock() + return val, nil + } + r.mu.Unlock() + + value, err := readTreehash(ctx, r.storage, build, r.emitter, r.op) + if err != nil { + return "", err + } + + r.mu.Lock() + r.cache[key] = value + r.mu.Unlock() + return value, nil } // GetChangedTargets returns the changed targets between two revisions. If the @@ -114,16 +164,13 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str maxDist = request.GetOutputConfig().GetMaxDistance() } - // Read treehashes once for the entire request. These are passed through - // the request pipeline to avoid redundant storage reads. - treehash1, treehash2, err := readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), e, opGetChangedTargets) - if err != nil { - return fmt.Errorf("read revision treehash: %w", err) - } + // Create a request-scoped resolver to memoize treehash reads. + // This avoids redundant storage reads across the request pipeline. + resolver := newTreehashResolver(c.storage, e, opGetChangedTargets) // Fast path: stream a previously computed result straight from cache. if !request.GetBypassCache() { - served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start, treehash1, treehash2) + served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start, resolver) if err != nil { return fmt.Errorf("serve from cache: %w", err) } @@ -133,7 +180,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str } // Fetch both revisions' target graphs concurrently. - firstGraph, secondGraph, err := c.fetchTargetGraphs(ctx, e, logger, request, treehash1, treehash2) + firstGraph, secondGraph, err := c.fetchTargetGraphs(ctx, e, logger, request, resolver) if err != nil { return fmt.Errorf("fetch target graphs: %w", err) } @@ -150,7 +197,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str } // Cache the computed result concurrently so it doesn't block the stream send. - c.cacheComparedTargets(logger, request, changedTargetsResponses, treehash1, treehash2) + c.cacheComparedTargets(logger, request, changedTargetsResponses, resolver) sendStart := time.Now() if err := sendTrimmedChangedTargets(stream, changedTargetsResponses, maxDist, request.GetOutputConfig()); err != nil { @@ -171,19 +218,15 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // - (true, nil) when a cached result was found and fully sent to the client; // - (false, nil) on a cache miss or a corrupt blob — the caller should recompute; // - (false, err) on an infra failure or a client disconnect that aborts the request. -// -// If treehash1 and treehash2 are provided (non-empty), they are used instead of -// reading from storage. This allows the caller to reuse treehash values already -// read earlier in the request. -func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time, treehash1, treehash2 string) (bool, error) { +func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time, resolver *treehashResolver) (bool, error) { cacheStart := time.Now() - if treehash1 == "" || treehash2 == "" { - // Treehashes not provided by caller; read them now. - var err error - treehash1, treehash2, err = readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), e, opGetChangedTargets) - if err != nil { - return false, fmt.Errorf("read revision treehash: %w", err) - } + treehash1, err := resolver.resolve(ctx, request.GetFirstRevision()) + if err != nil { + return false, fmt.Errorf("read revision treehash: %w", err) + } + treehash2, err := resolver.resolve(ctx, request.GetSecondRevision()) + if err != nil { + return false, fmt.Errorf("read revision treehash: %w", err) } if treehash1 == "" || treehash2 == "" { return false, nil @@ -252,17 +295,19 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metric // original failure is returned. A client disconnect surfaces as a user-cancelled // error. A graph stored as a TGB blob comes back as its undrained reader; a // gob-era graph is drained into chunks here, inside the concurrent fetch. -func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, treehash1, treehash2 string) (fetchedGraph, fetchedGraph, error) { +func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, resolver *treehashResolver) (fetchedGraph, fetchedGraph, error) { jobs := make([]*job, 2) for i := 0; i < 2; i++ { // create independent contexts for each job; if one of the jobs fails, the other one should be cancelled to save resources and improve latency ctxNew, cancelNew := context.WithCancelCause(ctx) defer cancelNew(nil) - th := treehash1 - if i == 1 { - th = treehash2 + var revision *pb.BuildDescription + if i == 0 { + revision = request.GetFirstRevision() + } else { + revision = request.GetSecondRevision() } - jobs[i] = &job{ctx: ctxNew, cancel: cancelNew, treehash: th} + jobs[i] = &job{ctx: ctxNew, cancel: cancelNew, revision: revision} } // Start jobs for both revisions. Success or failure, the result will report to the results channel. @@ -283,12 +328,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, results <- graphResult{order: idx, err: fmt.Errorf("panic in graph fetch: %v", r)} } }() - var revision *pb.BuildDescription - if idx == 0 { - revision = request.GetFirstRevision() - } else { - revision = request.GetSecondRevision() - } + revision := jobs[idx].revision entityBuild, err := mapper.ProtoToBuildDescription(revision) if err != nil { results <- graphResult{order: idx, err: fmt.Errorf("convert build description: %w", err)} @@ -299,7 +339,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, ExcludeFilesRegex: request.GetRequestOptions().GetExtraExcludeFilesRegex(), BypassCache: request.GetBypassCache(), } - graphReader, err := c.getGraph(jobs[idx].ctx, e, entityReq, jobs[idx].treehash) + graphReader, err := c.getGraph(jobs[idx].ctx, e, entityReq, resolver, revision) if err != nil || graphReader == nil { results <- graphResult{order: idx, err: err} return @@ -391,7 +431,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, // a fire-and-forget goroutine so it does not block the stream send. The responses // is only read (never mutated) by the goroutine and the foreground send, so // concurrent access is safe; the caller must not mutate it. This is best effort. -func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetChangedTargetsRequest, responses []entity.GetChangedTargetsResponse, treehash1, treehash2 string) { +func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetChangedTargetsRequest, responses []entity.GetChangedTargetsResponse, resolver *treehashResolver) { go func() { // Use c.appCtx directly: the cache write is fire-and-forget and must // outlive the request (so a client disconnect doesn't abort it) but @@ -400,8 +440,17 @@ func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetCha // is cancelled on shutdown. Per-operation deadlines are the storage // backend's responsibility — the controller is backend-agnostic and // must not encode any one implementation's I/O budget. - // The treehash values are passed by the caller (already read during - // the request) to avoid redundant storage reads. + // Use the resolver to get treehash values (memoized from request). + treehash1, err := resolver.resolve(c.appCtx, request.GetFirstRevision()) + if err != nil { + logger.Warn("GetChangedTargets: skipping cache write, failed to read revision treehash", zap.Error(err)) + return + } + treehash2, err := resolver.resolve(c.appCtx, request.GetSecondRevision()) + if err != nil { + logger.Warn("GetChangedTargets: skipping cache write, failed to read revision treehash", zap.Error(err)) + return + } if treehash1 != "" && treehash2 != "" { cacheKey := cachekey.GetComparedTargetsCachePath(request.GetFirstRevision().GetRemote(), treehash1, treehash2, request.GetRequestOptions().GetExtraExcludeFilesRegex()) if writeErr := storage.WriteChangedTargetsStream(c.appCtx, c.storage, cacheKey, responses); writeErr != nil { diff --git a/controller/getchangedtargets_test.go b/controller/getchangedtargets_test.go index dbaacd76..037f790a 100644 --- a/controller/getchangedtargets_test.go +++ b/controller/getchangedtargets_test.go @@ -358,13 +358,12 @@ func TestGetChangedTargets_TreehashReadError(t *testing.T) { storagemock := storagemock.NewMockStorage(ctrl) // A non-NotFound storage error on a treehash read must surface as a failed - // request rather than be silently treated as a cache miss. Both revision - // treehashes are read in parallel, so two Get calls happen; the handler - // returns the first failure (and drops the cancelled sibling's error) - // before any graph fetch happens. + // request rather than be silently treated as a cache miss. The resolver + // reads treehashes sequentially, so the first error is returned and the + // second read is not attempted. injected := errors.New("storage exploded") storagemock.EXPECT().Get(gomock.Any(), gomock.Any()). - Return(storage.DownloadResponse{}, injected).Times(2) + Return(storage.DownloadResponse{}, injected).Times(1) c := NewController(context.Background(), Params{ Logger: zap.NewNop(), @@ -1407,8 +1406,9 @@ func TestServeChangedTargetsFromCache(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.storage = st stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) + resolver := newTreehashResolver(c.storage, c.emitter, opGetChangedTargets) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), "", "") + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver) require.NoError(t, err) assert.False(t, served, "a cache miss must not be served") }) @@ -1445,9 +1445,10 @@ func TestServeChangedTargetsFromCache(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.storage = st stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) + resolver := newTreehashResolver(c.storage, c.emitter, opGetChangedTargets) // No Send expectation: a corrupt blob must not send anything to the client. - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), "", "") + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver) require.NoError(t, err) assert.False(t, served, "a corrupt blob must trigger recompute, not a partial send") }) @@ -1480,8 +1481,9 @@ func TestServeChangedTargetsFromCache(t *testing.T) { c.storage = st stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2) + resolver := newTreehashResolver(c.storage, c.emitter, opGetChangedTargets) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), "", "") + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver) require.NoError(t, err) assert.True(t, served, "a clean cache hit must be served") }) @@ -1507,8 +1509,9 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch + resolver := newTreehashResolver(storage.NewMemoryStorage(), c.emitter, opGetChangedTargets) - first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") + first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), resolver) require.NoError(t, err) require.Len(t, first.chunks, 1) require.Len(t, second.chunks, 1) @@ -1531,8 +1534,9 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch + resolver := newTreehashResolver(storage.NewMemoryStorage(), c.emitter, opGetChangedTargets) - first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") + first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), resolver) require.Error(t, err) assert.ErrorIs(t, err, injected) assert.Zero(t, first) @@ -1552,8 +1556,9 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch + resolver := newTreehashResolver(storage.NewMemoryStorage(), c.emitter, opGetChangedTargets) - _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") + _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), resolver) require.Error(t, err) }) @@ -1567,8 +1572,9 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch + resolver := newTreehashResolver(storage.NewMemoryStorage(), c.emitter, opGetChangedTargets) - _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), "", "") + _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest(), resolver) require.Error(t, err) }) } diff --git a/controller/gettargetgraph.go b/controller/gettargetgraph.go index 257ea6c4..ec6ac553 100644 --- a/controller/gettargetgraph.go +++ b/controller/gettargetgraph.go @@ -56,7 +56,7 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb if err != nil { return tangoerrors.NewUser(fmt.Errorf("convert get target graph request: %w", err)) } - graphReader, err := c.getGraph(ctx, e, entityReq, "") + graphReader, err := c.getGraph(ctx, e, entityReq, nil, request.GetBuildDescription()) if err != nil { return fmt.Errorf("get graph: %w", err) } @@ -96,18 +96,24 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb // entries store the full payload and stripping happens at send time, so // letting an orchestrator see it could poison the shared cache with // stripped graphs. -// If treehash is provided (non-empty), it is used instead of reading from storage. -func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entity.GetTargetGraphRequest, treehash string) (storage.GraphReader, error) { +// If resolver is provided, it is used to resolve the treehash for the build description. +func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entity.GetTargetGraphRequest, resolver *treehashResolver, revision *pb.BuildDescription) (storage.GraphReader, error) { start := time.Now() logger := c.logger.With( zap.Any("build_description", req.Build), ) if !req.BypassCache { var treehashValue string - if treehash != "" { - // Use the pre-read treehash value passed by the caller. - treehashValue = treehash - logger.Debug("getGraph: using pre-read treehash") + if resolver != nil { + // Use the resolver to memoize treehash reads. + var err error + treehashValue, err = resolver.resolve(ctx, revision) + if err != nil { + return nil, fmt.Errorf("get treehash: %w", err) + } + if treehashValue != "" { + logger.Debug("getGraph: using pre-read treehash") + } } else { // Look up the git treehash based on cache path treehashCachePath := cachekey.GetTreehashCachePath(req.Build) From 055f6e12a0a1ea21742a3ef754a38e7436206237 Mon Sep 17 00:00:00 2001 From: Arjun Mehta Date: Fri, 28 Aug 2026 04:49:53 +0000 Subject: [PATCH 3/3] refactor(controller): preserve parallel treehash resolution Keep treehash resolution request-scoped while preserving the concurrent lookup behavior used by GetChangedTargets. Memoize successful treehash values without caching errors, and reuse the resolved values across the request pipeline. --- controller/getchangedtargets.go | 77 ++++++++++++++++++++++------ controller/getchangedtargets_test.go | 18 ++++--- 2 files changed, 70 insertions(+), 25 deletions(-) diff --git a/controller/getchangedtargets.go b/controller/getchangedtargets.go index 597ca904..fa145c5e 100644 --- a/controller/getchangedtargets.go +++ b/controller/getchangedtargets.go @@ -129,6 +129,45 @@ func (r *treehashResolver) resolve(ctx context.Context, build *pb.BuildDescripti return value, nil } +// resolveParallel resolves both treehashes concurrently. +// It memoizes successful reads; errors are not cached. +// Returns (treehash1, treehash2, error). +func (r *treehashResolver) resolveParallel(ctx context.Context, first, second *pb.BuildDescription) (string, string, error) { + type result struct { + idx int + hash string + err error + } + + descs := [2]*pb.BuildDescription{first, second} + results := make(chan result, 2) + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + for i, desc := range descs { + go func(idx int, d *pb.BuildDescription) { + hash, err := r.resolve(ctx, d) + results <- result{idx: idx, hash: hash, err: err} + }(i, desc) + } + + var hashes [2]string + var firstErr error + for range descs { + res := <-results + hashes[res.idx] = res.hash + if res.err != nil && firstErr == nil { + firstErr = res.err + cancel() + } + } + if firstErr != nil { + return "", "", firstErr + } + return hashes[0], hashes[1], nil +} + // GetChangedTargets returns the changed targets between two revisions. If the // client disconnects, the stream's context is cancelled and the function // returns with context.Canceled. @@ -168,14 +207,24 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // This avoids redundant storage reads across the request pipeline. resolver := newTreehashResolver(c.storage, e, opGetChangedTargets) - // Fast path: stream a previously computed result straight from cache. - if !request.GetBypassCache() { - served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start, resolver) - if err != nil { - return fmt.Errorf("serve from cache: %w", err) - } - if served { - return nil + // Read both treehashes concurrently to populate the resolver cache. + // This preserves the original concurrent behavior from readTreehashParallel. + treehash1, treehash2, err := resolver.resolveParallel(ctx, request.GetFirstRevision(), request.GetSecondRevision()) + if err != nil { + return fmt.Errorf("read revision treehash: %w", err) + } + if treehash1 == "" || treehash2 == "" { + // One or both treehashes missing; skip cache and recompute. + } else { + // Fast path: stream a previously computed result straight from cache. + if !request.GetBypassCache() { + served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start, resolver, treehash1, treehash2) + if err != nil { + return fmt.Errorf("serve from cache: %w", err) + } + if served { + return nil + } } } @@ -218,16 +267,10 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // - (true, nil) when a cached result was found and fully sent to the client; // - (false, nil) on a cache miss or a corrupt blob — the caller should recompute; // - (false, err) on an infra failure or a client disconnect that aborts the request. -func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time, resolver *treehashResolver) (bool, error) { +// +// treehash1 and treehash2 are pre-resolved treehash values (may be empty if cache miss). +func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time, resolver *treehashResolver, treehash1, treehash2 string) (bool, error) { cacheStart := time.Now() - treehash1, err := resolver.resolve(ctx, request.GetFirstRevision()) - if err != nil { - return false, fmt.Errorf("read revision treehash: %w", err) - } - treehash2, err := resolver.resolve(ctx, request.GetSecondRevision()) - if err != nil { - return false, fmt.Errorf("read revision treehash: %w", err) - } if treehash1 == "" || treehash2 == "" { return false, nil } diff --git a/controller/getchangedtargets_test.go b/controller/getchangedtargets_test.go index 037f790a..e5d61af2 100644 --- a/controller/getchangedtargets_test.go +++ b/controller/getchangedtargets_test.go @@ -359,11 +359,13 @@ func TestGetChangedTargets_TreehashReadError(t *testing.T) { storagemock := storagemock.NewMockStorage(ctrl) // A non-NotFound storage error on a treehash read must surface as a failed // request rather than be silently treated as a cache miss. The resolver - // reads treehashes sequentially, so the first error is returned and the - // second read is not attempted. + // reads treehashes concurrently, so both reads are attempted; the first + // error is returned and the sibling is cancelled. injected := errors.New("storage exploded") + // Both reads are attempted concurrently; one will fail and the other will be cancelled. + // The exact number of calls depends on scheduling, so we accept at least 1 call. storagemock.EXPECT().Get(gomock.Any(), gomock.Any()). - Return(storage.DownloadResponse{}, injected).Times(1) + Return(storage.DownloadResponse{}, injected).MinTimes(1) c := NewController(context.Background(), Params{ Logger: zap.NewNop(), @@ -1399,16 +1401,16 @@ func TestServeChangedTargetsFromCache(t *testing.T) { t.Run("cache miss returns not-served, no error", func(t *testing.T) { ctrl := gomock.NewController(t) st := storagemock.NewMockStorage(ctrl) - // Both treehash reads miss, so the cache path is skipped entirely. + // Cache miss for compared-targets, so the cache path is skipped entirely. st.EXPECT().Get(gomock.Any(), gomock.Any()). - Return(storage.DownloadResponse{}, storage.NewNotFoundError("missing")).Times(2) + Return(storage.DownloadResponse{}, storage.NewNotFoundError("missing")).Times(1) c := newTestController(zaptest.NewLogger(t)) c.storage = st stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) resolver := newTreehashResolver(c.storage, c.emitter, opGetChangedTargets) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver, "treehash1", "treehash2") require.NoError(t, err) assert.False(t, served, "a cache miss must not be served") }) @@ -1448,7 +1450,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { resolver := newTreehashResolver(c.storage, c.emitter, opGetChangedTargets) // No Send expectation: a corrupt blob must not send anything to the client. - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver, "", "") require.NoError(t, err) assert.False(t, served, "a corrupt blob must trigger recompute, not a partial send") }) @@ -1483,7 +1485,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2) resolver := newTreehashResolver(c.storage, c.emitter, opGetChangedTargets) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now(), resolver, "treehash1", "treehash2") require.NoError(t, err) assert.True(t, served, "a clean cache hit must be served") })