From 018ae0269618c1857f05b14c9a082bd5cc1cf87c Mon Sep 17 00:00:00 2001 From: Xiaoyang Tan Date: Tue, 17 Feb 2026 15:01:41 -0800 Subject: [PATCH 1/2] fix changed targets --- core/bazel/query.go | 7 +- core/controller/getchangedtargets.go | 55 +++++----- core/controller/getchangedtargets_test.go | 120 ++++++++++++++++++++-- example/client/client.go | 5 + 4 files changed, 149 insertions(+), 38 deletions(-) diff --git a/core/bazel/query.go b/core/bazel/query.go index 64607b23..9efc27db 100644 --- a/core/bazel/query.go +++ b/core/bazel/query.go @@ -83,12 +83,7 @@ func (b *BazelClient) executeQueryInternal(ctx context.Context, query string, st b.logger.Error("Error in stream processing: %v", streamErr) return nil, streamErr } - // TODO: Remove these placeholder logs - b.logger.Info("\nParsed targets (%d):\n", len(queryResults.Target)) - b.logger.Info("STDOUT: %s", stdoutBuf.String()) - if stderrBuf.Len() > 0 { - b.logger.Debugf("STDERR: %s", stderrBuf.String()) - } + b.logger.Debugf("Parsed %d targets from bazel query", len(queryResults.Target)) return queryResults, nil } diff --git a/core/controller/getchangedtargets.go b/core/controller/getchangedtargets.go index 146cc91d..750981cd 100644 --- a/core/controller/getchangedtargets.go +++ b/core/controller/getchangedtargets.go @@ -49,10 +49,9 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // Start jobs for both revisions. Success or failure, the result will report to the results channel. type graphResult struct { // order is 0 or 1, 0 is the base (first) revision, 1 is the target (second) revision - order int - // TODO: pb.GetTargetGraphResponse is a stream, so most likely we can't use GetTargetGraphResponse as a return type and we'll want to read it fully before joining the threads - graph *pb.GetTargetGraphResponse - err error + order int + chunks []*pb.GetTargetGraphResponse + err error } results := make(chan graphResult, len(jobs)) @@ -71,11 +70,25 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str return } if graphReader == nil { - results <- graphResult{order: idx, err: nil} + results <- graphResult{order: idx} return } - graph, err := graphReader.Read() - results <- graphResult{order: idx, graph: graph, err: err} + defer graphReader.Close() + + // Read all chunks from the stream + var chunks []*pb.GetTargetGraphResponse + for { + chunk, err := graphReader.Read() + if err == io.EOF { + break + } + if err != nil { + results <- graphResult{order: idx, err: err} + return + } + chunks = append(chunks, chunk) + } + results <- graphResult{order: idx, chunks: chunks} }(i) } @@ -83,16 +96,8 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str for range jobs { select { case res := <-results: - if res.graph != nil { - jobs[res.order].graphStreamChunks = append(jobs[res.order].graphStreamChunks, res.graph) - } - if res.graph == nil { - jobs[res.order].completed = true - } - if res.err == io.EOF { - res.err = nil - jobs[res.order].completed = true - } + jobs[res.order].graphStreamChunks = res.chunks + jobs[res.order].completed = true if res.err != nil { jobs[res.order].err = res.err @@ -209,7 +214,7 @@ func (c *controller) compareTargetGraphs(ctx context.Context, firstGraph, second initial := pb.CHANGE_TYPE_INDIRECT // If we know the source file rule type, classify changes accordingly. // Otherwise, leave as UNSPECIFIED. - // check if the target is a source file, if so, it is a direct change + // check if the target is a source file, if so, it is a direct change isSource := newT.GetRuleType() == sourceFileRuleTypeID && sourceFileRuleTypeID != -1 if isSource { initial = pb.CHANGE_TYPE_DIRECT @@ -240,8 +245,8 @@ func (c *controller) compareTargetGraphs(ctx context.Context, firstGraph, second ) changedByName[name] = &pb.ChangedTarget{ ChangeType: initial, - OldTarget: oldTarget, - NewTarget: newTarget, + OldTarget: oldTarget, + NewTarget: newTarget, } } @@ -287,11 +292,11 @@ func (c *controller) compareTargetGraphs(ctx context.Context, firstGraph, second // 5) Construct canonical metadata and emit responses. meta := &pb.Metadata{ - TargetIdMapping: targetMapper.Invert(), - RuleTypeMapping: ruleTypeMapper.Invert(), - TagMapping: tagMapper.Invert(), - AttributeNameMapping: attrNameMapper.Invert(), - AttributeStringValueMapping: attrValMapper.Invert(), + TargetIdMapping: targetMapper.Invert(), + RuleTypeMapping: ruleTypeMapper.Invert(), + TagMapping: tagMapper.Invert(), + AttributeNameMapping: attrNameMapper.Invert(), + AttributeStringValueMapping: attrValMapper.Invert(), } // Emit changes and metadata as separate responses. diff --git a/core/controller/getchangedtargets_test.go b/core/controller/getchangedtargets_test.go index 2d7a658e..d5dc45c3 100644 --- a/core/controller/getchangedtargets_test.go +++ b/core/controller/getchangedtargets_test.go @@ -4,17 +4,19 @@ import ( "bytes" "context" "errors" + "fmt" "io" + "strings" "testing" + gogio "github.com/gogo/protobuf/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/uber/tango/core/storage" storagemock "github.com/uber/tango/core/storage/storagemock" orchestratormock "github.com/uber/tango/orchestrator/orchestratormock" pb "github.com/uber/tango/tangopb" tangomock "github.com/uber/tango/tangopb/tangopbmock" - gogio "github.com/gogo/protobuf/io" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" "go.uber.org/mock/gomock" "go.uber.org/zap" "go.uber.org/zap/zaptest" @@ -180,7 +182,7 @@ func TestGetChangedTargets_StreamSendError(t *testing.T) { assert.Error(t, err) } -func TestGetChangedTargets_Success(t *testing.T) { +func TestGetChangedTargets_singleChunk(t *testing.T) { ctrl := gomock.NewController(t) stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) @@ -213,6 +215,110 @@ func TestGetChangedTargets_Success(t *testing.T) { assert.NoError(t, err) } +func TestGetChangedTargets_streamChunks(t *testing.T) { + ctrl := gomock.NewController(t) + stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) + + 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 + w1 := gogio.NewDelimitedWriter(&buf1) + w1.WriteMsg(&pb.GetTargetGraphResponse{ + Item: &pb.GetTargetGraphResponse_Targets{ + Targets: &pb.OptimizedTargets{ + Targets: []*pb.OptimizedTarget{ + {Id: 1, Hash: "h1", RuleType: 100}, + {Id: 2, Hash: "h2-old", RuleType: 200}, + }, + }, + }, + }) + w1.WriteMsg(&pb.GetTargetGraphResponse{ + Item: &pb.GetTargetGraphResponse_Metadata{ + Metadata: &pb.Metadata{ + TargetIdMapping: map[int32]string{1: "//app:target1", 2: "//app:target2"}, + RuleTypeMapping: map[int32]string{100: "go_library", 200: "go_binary"}, + }, + }, + }) + graph1Bytes := buf1.Bytes() + + // Build second revision graph - target2 has different hash + var buf2 bytes.Buffer + w2 := gogio.NewDelimitedWriter(&buf2) + w2.WriteMsg(&pb.GetTargetGraphResponse{ + Item: &pb.GetTargetGraphResponse_Targets{ + Targets: &pb.OptimizedTargets{ + Targets: []*pb.OptimizedTarget{ + {Id: 1, Hash: "h1", RuleType: 100}, + {Id: 2, Hash: "h2-new", RuleType: 200}, // changed hash + }, + }, + }, + }) + w2.WriteMsg(&pb.GetTargetGraphResponse{ + Item: &pb.GetTargetGraphResponse_Metadata{ + Metadata: &pb.Metadata{ + TargetIdMapping: map[int32]string{1: "//app:target1", 2: "//app:target2"}, + RuleTypeMapping: map[int32]string{100: "go_library", 200: "go_binary"}, + }, + }, + }) + graph2Bytes := buf2.Bytes() + + // Each revision needs: treehash lookup + graph lookup + storagemock.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, req storage.DownloadRequest) (*storage.DownloadResponse, error) { + switch { + case strings.Contains(req.Key, "sha1"): + return &storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader([]byte("treehash1")))}, nil + case strings.Contains(req.Key, "sha2"): + return &storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader([]byte("treehash2")))}, nil + case strings.Contains(req.Key, "treehash1"): + return &storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader(graph1Bytes))}, nil + case strings.Contains(req.Key, "treehash2"): + return &storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader(graph2Bytes))}, nil + default: + return nil, fmt.Errorf("unexpected key: %s", req.Key) + } + }).Times(4) + + c := NewController(Params{ + Logger: zaptest.NewLogger(t), + Storage: storagemock, + Orchestrator: orchestratormock.NewMockOrchestrator(ctrl), + }) + + request := &pb.GetChangedTargetsRequest{ + FirstRevision: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha1"}, + SecondRevision: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha2"}, + } + + err := c.GetChangedTargets(request, stream) + require.NoError(t, err) + + // xactly 2 responses: ChangedTargets + Metadata + require.Len(t, sentResponses, 2) + changedTargets := sentResponses[0].GetChangedTargets() + metadata := sentResponses[1].GetMetadata() + + // Verify target2 is detected as changed (hash changed from h2-old to h2-new) + require.Len(t, changedTargets.GetChangedTargets(), 1, "should detect 1 changed target") + changed := changedTargets.GetChangedTargets()[0] + assert.Equal(t, "h2-old", changed.GetOldTarget().GetHash()) + assert.Equal(t, "h2-new", changed.GetNewTarget().GetHash()) + + targetID := changed.GetNewTarget().GetId() + assert.Equal(t, "//app:target2", metadata.GetTargetIdMapping()[targetID]) +} + func TestCompareTargetGraphs_NewTarget_CanonicalIDs(t *testing.T) { c := &controller{logger: zaptest.NewLogger(t)} @@ -275,7 +381,7 @@ func TestCompareTargetGraphs_SourceFileDirectAndPropagation(t *testing.T) { Item: &pb.GetTargetGraphResponse_Targets{ Targets: &pb.OptimizedTargets{ Targets: []*pb.OptimizedTarget{ - {Id: 1, Hash: "h1", RuleType: 100}, // "source file" + {Id: 1, Hash: "h1", RuleType: 100}, // "source file" {Id: 2, Hash: "h1", RuleType: 200, DirectDependencies: []int32{1}}, // "rule" }, }, @@ -302,7 +408,7 @@ func TestCompareTargetGraphs_SourceFileDirectAndPropagation(t *testing.T) { Item: &pb.GetTargetGraphResponse_Targets{ Targets: &pb.OptimizedTargets{ Targets: []*pb.OptimizedTarget{ - {Id: 11, Hash: "h2", RuleType: 101}, // "source file" + {Id: 11, Hash: "h2", RuleType: 101}, // "source file" {Id: 22, Hash: "h2", RuleType: 201, DirectDependencies: []int32{11}}, // "rule" }, }, @@ -673,7 +779,7 @@ func TestCompareTargetGraphs_IndirectWhenOnlyHashChanged(t *testing.T) { Id: 2, Hash: "h2", // Changed RuleType: 201, - DirectDependencies: []int32{20}, // Same dep (//app:A) + DirectDependencies: []int32{20}, // Same dep (//app:A) Attributes: map[int32]int32{2: 20}, // Same attribute }, {Id: 20, Hash: "h2", RuleType: 201}, // Dep A hash changed diff --git a/example/client/client.go b/example/client/client.go index 940d7ddb..7e924198 100644 --- a/example/client/client.go +++ b/example/client/client.go @@ -189,6 +189,11 @@ func callGetChangedTargets(ctx context.Context, client pb.TangoYARPCClient, logg switch x := msg.Item.(type) { case *pb.GetChangedTargetsResponse_ChangedTargets: fmt.Printf("Received changed targets packet with %d targets\n", len(x.ChangedTargets.GetChangedTargets())) + json, err := json.Marshal(x.ChangedTargets) + if err != nil { + return fmt.Errorf("marshal changed targets: %w", err) + } + fmt.Printf("ChangedTargets: %s\n", string(json)) case *pb.GetChangedTargetsResponse_Metadata: // unmarshal response to json json, err := json.Marshal(x.Metadata) From be33c1b053ba23a0d715337d91bdde00b30f1e25 Mon Sep 17 00:00:00 2001 From: Xiaoyang Tan Date: Tue, 17 Feb 2026 15:46:01 -0800 Subject: [PATCH 2/2] fix --- core/controller/getchangedtargets.go | 25 ++++++++----- core/controller/getchangedtargets_test.go | 45 ++++++----------------- 2 files changed, 26 insertions(+), 44 deletions(-) diff --git a/core/controller/getchangedtargets.go b/core/controller/getchangedtargets.go index 750981cd..a99ebab8 100644 --- a/core/controller/getchangedtargets.go +++ b/core/controller/getchangedtargets.go @@ -65,14 +65,11 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str revision = request.GetSecondRevision() } graphReader, err := c.getGraph(jobs[idx].ctx, revision, request.GetOutputConfig()) - if err != nil { + // something's wrong with getGraph + if err != nil || graphReader == nil { results <- graphResult{order: idx, err: err} return } - if graphReader == nil { - results <- graphResult{order: idx} - return - } defer graphReader.Close() // Read all chunks from the stream @@ -98,15 +95,23 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str case res := <-results: jobs[res.order].graphStreamChunks = res.chunks jobs[res.order].completed = true - if res.err != nil { - jobs[res.order].err = res.err + jobs[res.order].err = res.err + if res.err == io.EOF { + jobs[res.order].err = nil + } + if res.chunks == nil && res.err == nil { + jobs[res.order].err = errors.New("no chunks returned") + } - // one of the computations failed, if the other one has not completed yet, cancel it and wait for the result to come in, which would be a context cancelled result then + // one of the computations failed, if the other one has not + // completed yet, cancel it and wait for the result to come in, + // which would be a context cancelled result then + if res.err != nil { other := (res.order + 1) % 2 if !jobs[other].completed { jobs[other].cancel() - - // explicitly mark that this job is cancelled, so we can ignore its error later + // explicitly mark that this job is cancelled, so we can + // ignore its error later jobs[other].cancelled = true } } diff --git a/core/controller/getchangedtargets_test.go b/core/controller/getchangedtargets_test.go index d5dc45c3..1d83ba39 100644 --- a/core/controller/getchangedtargets_test.go +++ b/core/controller/getchangedtargets_test.go @@ -165,7 +165,17 @@ func TestGetChangedTargets_StreamSendError(t *testing.T) { stream.EXPECT().Send(gomock.Any()).Return(errors.New("send error")) storagemock := storagemock.NewMockStorage(ctrl) - storagemock.EXPECT().Get(gomock.Any(), gomock.Any()).Return(&storage.DownloadResponse{ReadCloser: nil}, nil).AnyTimes() + + var buf bytes.Buffer + gogio.NewDelimitedWriter(&buf).WriteMsg(&pb.GetTargetGraphResponse{ + Item: &pb.GetTargetGraphResponse_Targets{Targets: &pb.OptimizedTargets{}}, + }) + storagemock.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, req storage.DownloadRequest) (*storage.DownloadResponse, error) { + if strings.Contains(req.Key, "th") { + return &storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader(buf.Bytes()))}, nil + } + return &storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader([]byte("th")))}, nil + }).AnyTimes() c := NewController(Params{ Logger: zaptest.NewLogger(t), @@ -182,39 +192,6 @@ func TestGetChangedTargets_StreamSendError(t *testing.T) { assert.Error(t, err) } -func TestGetChangedTargets_singleChunk(t *testing.T) { - ctrl := gomock.NewController(t) - stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) - - stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2) - storagemock := storagemock.NewMockStorage(ctrl) - // Prepare graph bytes to be returned for graph fetches - graph := pb.GetTargetGraphResponse{Item: &pb.GetTargetGraphResponse_Targets{Targets: &pb.OptimizedTargets{}}} - var buf bytes.Buffer - err := gogio.NewDelimitedWriter(&buf).WriteMsg(&graph) - require.NoError(t, err) - // Controller.getGraph performs two storage lookups per revision: - // 1) treehash cache -> returns bytes (content not important) - // 2) graph by treehash -> returns marshaled graph - // We set four Get calls total; concurrency means order may vary, but returning either is acceptable. - storagemock.EXPECT().Get(gomock.Any(), gomock.Any()).Return(&storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader([]byte("th")))}, nil).Times(2) - storagemock.EXPECT().Get(gomock.Any(), gomock.Any()).Return(&storage.DownloadResponse{ReadCloser: io.NopCloser(bytes.NewReader(buf.Bytes()))}, nil).Times(2) - orchestrator := orchestratormock.NewMockOrchestrator(ctrl) - c := NewController(Params{ - Logger: zaptest.NewLogger(t), - Storage: storagemock, - Orchestrator: orchestrator, - }) - - request := &pb.GetChangedTargetsRequest{ - FirstRevision: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha1"}, - SecondRevision: &pb.BuildDescription{Remote: "repo:go-code", BaseSha: "sha2"}, - } - - err = c.GetChangedTargets(request, stream) - assert.NoError(t, err) -} - func TestGetChangedTargets_streamChunks(t *testing.T) { ctrl := gomock.NewController(t) stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)