fix changed targets streaming

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
 }
 

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))
 
@@ -66,42 +65,53 @@ 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, err: nil}
-				return
+			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)
 			}
-			graph, err := graphReader.Read()
-			results <- graphResult{order: idx, graph: graph, err: err}
+			results <- graphResult{order: idx, chunks: chunks}
 		}(i)
 	}
 
 	// Wait for both results to complete, either successfully or with an error.
 	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
-			}
+			jobs[res.order].graphStreamChunks = res.chunks
+			jobs[res.order].completed = true
+			jobs[res.order].err = res.err
 			if res.err == io.EOF {
-				res.err = nil
-				jobs[res.order].completed = true
+				jobs[res.order].err = nil
+			}
+			if res.chunks == nil && res.err == nil {
+				jobs[res.order].err = errors.New("no chunks returned")
 			}
-			if res.err != nil {
-				jobs[res.order].err = res.err
 
-				// 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
 				}
 			}
@@ -209,7 +219,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 +250,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 +297,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.

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"
@@ -163,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),
@@ -180,37 +192,108 @@ func TestGetChangedTargets_StreamSendError(t *testing.T) {
 	assert.Error(t, err)
 }
 
-func TestGetChangedTargets_Success(t *testing.T) {
+func TestGetChangedTargets_streamChunks(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl)
 
-	stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2)
+	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)
-	// 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)
+
+	// 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: orchestrator,
+		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)
-	assert.NoError(t, err)
+	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) {
@@ -275,7 +358,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 +385,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 +756,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

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)