feat(conflict): Record target labels in the dictionary when signing a batch

submitqueue/extension/conflict/tango/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
         "//submitqueue/extension/conflict:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
     ],
 )
 
@@ -45,5 +46,7 @@ go_test(
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+        "@org_uber_go_zap//zaptest/observer:go_default_library",
     ],
 )

submitqueue/extension/conflict/tango/README.md

@@ -10,6 +10,7 @@ A batch's signature is the sorted set of its affected targets, each as a 64-bit
 
 - Only the batch being analyzed is ever sent to Tango, and only when it has no recorded signature. Redelivery finds the record the earlier attempt created and does not call Tango again.
 - The new batch is signed even when nothing is in flight, so later batches find its record.
+- After the attempt that created a signature succeeds, it records each target's label in the dictionary in the background, off the request path, so an ID can be read back as a label. This is best-effort: a failed write is logged and skipped, because conflicts are decided by ID alone and a later batch that touches the label writes it again. On a hash collision the first label recorded stays.
 - An in-flight batch is read from the cache, then the store, and never from Tango. A missing record or a store error fails the analysis (`ErrNoSignature` for the missing record), because a signature computed later could differ from the one the batch was admitted with.
 - The analyzer fails closed on any error.
 - A queue must have no in-flight batches when it is switched to this analyzer, because batches admitted earlier have no signature.

submitqueue/extension/conflict/tango/analyzer.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"time"
 
+	"go.uber.org/zap"
+
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/conflict"
@@ -37,6 +39,10 @@ type Params struct {
 	// analyzer serves.
 	Signatures SignatureStore
 
+	// Dictionary records the label behind each target ID, in the background and
+	// best-effort. It must be bound to the queue the analyzer serves.
+	Dictionary DictionaryStore
+
 	// Cache holds recently used signatures and may be shared across queues.
 	Cache *SignatureCache
 
@@ -49,6 +55,9 @@ type Params struct {
 
 	// Changes resolves a batch into the changes it contains.
 	Changes changeset.Resolver
+
+	// Logger reports best-effort failures. It defaults to a no-op logger.
+	Logger *zap.SugaredLogger
 }
 
 type analyzer struct {
@@ -60,6 +69,9 @@ type analyzer struct {
 // the analyzed batch's. It fails closed: any signature it cannot obtain fails
 // the analysis.
 func New(params Params) conflict.Analyzer {
+	if params.Logger == nil {
+		params.Logger = zap.NewNop().Sugar()
+	}
 	return &analyzer{Params: params}
 }
 
@@ -102,6 +114,18 @@ func (a *analyzer) recordedSignature(ctx context.Context, queue, batchID string)
 	return signature, true, nil
 }
 
+// recordLabels is best-effort: signatures compare IDs alone, and a label that
+// fails to record is written again by any later batch that touches it.
+func (a *analyzer) recordLabels(ctx context.Context, queue string, labels []hashedLabel) {
+	createdAt := time.Now().UnixMilli()
+	for _, l := range labels {
+		err := a.Dictionary.Create(ctx, TargetLabel{Queue: queue, ID: l.id, Label: l.label, CreatedAt: createdAt})
+		if err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
+			a.Logger.Warnw("failed to record target label", "queue", queue, "id", l.id, "label", l.label, "error", err)
+		}
+	}
+}
+
 // signNewBatch returns the batch's own record if an earlier attempt created
 // one; otherwise it creates the signature first, so racing attempts agree.
 func (a *analyzer) signNewBatch(ctx context.Context, batch entity.Batch) (TargetSignature, error) {
@@ -123,9 +147,8 @@ func (a *analyzer) signNewBatch(ctx context.Context, batch entity.Batch) (Target
 		return TargetSignature{}, fmt.Errorf("failed to get the changed targets: %w", err)
 	}
 
-	// TODO(tango): record the labels in the target dictionary before creating the signature.
 	// TODO(tango): add the destination of each new dependency edge to the signature, the RFC's known gap. The adapter must return direct dependencies first.
-	targets, _ := toSignatureTargets(affected)
+	targets, labels := toSignatureTargets(affected)
 	signature = TargetSignature{
 		Queue:        batch.Queue,
 		BatchID:      batch.ID,
@@ -145,6 +168,10 @@ func (a *analyzer) signNewBatch(ctx context.Context, batch entity.Batch) (Target
 		if !found {
 			return TargetSignature{}, fmt.Errorf("signature reported as existing but not found")
 		}
+	} else {
+		// Only the attempt that created the signature records its labels, off the
+		// request path and detached so a finished request does not cancel it.
+		go a.recordLabels(context.WithoutCancel(ctx), batch.Queue, labels)
 	}
 	a.Cache.put(signature)
 	return signature, nil

submitqueue/extension/conflict/tango/analyzer_test.go

@@ -17,11 +17,14 @@ package tango
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/uber/submitqueue/platform/base/change"
 	changesetmock "github.com/uber/submitqueue/submitqueue/core/changeset/mock"
@@ -40,22 +43,31 @@ var (
 
 type analyzerHarness struct {
 	signatures *MockSignatureStore
+	dictionary *MockDictionaryStore
+	logs       *observer.ObservedLogs
 	client     *MockChangedTargetsClient
 	baseErr    error
+	recorded   sync.WaitGroup
 	changes    *changesetmock.MockResolver
 	analyzer   conflict.Analyzer
 }
 
 func newAnalyzerHarness(t *testing.T) *analyzerHarness {
 	t.Helper()
 	ctrl := gomock.NewController(t)
+	core, logs := observer.New(zap.WarnLevel)
 	h := &analyzerHarness{
 		signatures: NewMockSignatureStore(ctrl),
+		dictionary: NewMockDictionaryStore(ctrl),
+		logs:       logs,
 		client:     NewMockChangedTargetsClient(ctrl),
 		changes:    changesetmock.NewMockResolver(ctrl),
 	}
+	t.Cleanup(h.recorded.Wait)
 	h.analyzer = New(Params{
 		Signatures:   h.signatures,
+		Dictionary:   h.dictionary,
+		Logger:       zap.New(core).Sugar(),
 		Cache:        NewSignatureCache(1 << 20),
 		Client:       h.client,
 		BaseRevision: func(context.Context, string) (string, error) { return analyzerTestBaseRevision, h.baseErr },
@@ -76,6 +88,11 @@ func analyzerTestNewSignature() TargetSignature {
 	}
 }
 
+func analyzerTestLabels() []hashedLabel {
+	_, labels := toSignatureTargets(analyzerTestAffected)
+	return labels
+}
+
 func analyzerTestSignature(batchID string, ids ...uint64) TargetSignature {
 	targets := make([]SignatureTarget, len(ids))
 	for i, id := range ids {
@@ -99,6 +116,14 @@ func (h *analyzerHarness) expectSigningNewBatch(createErr error) {
 		BaseRevision: analyzerTestBaseRevision,
 		ChangeURIs:   []string{"u1", "u2"},
 	}).Return(analyzerTestAffected, nil)
+	if createErr == nil {
+		for _, l := range analyzerTestLabels() {
+			h.recorded.Add(1)
+			h.dictionary.EXPECT().Create(gomock.Any(), gomock.Cond(func(got TargetLabel) bool {
+				return got.Queue == analyzerTestQueue && got.ID == l.id && got.Label == l.label && got.CreatedAt != 0
+			})).Do(func(context.Context, TargetLabel) { h.recorded.Done() }).Return(nil)
+		}
+	}
 	h.signatures.EXPECT().Create(gomock.Any(), gomock.Cond(func(got TargetSignature) bool {
 		stamped := got.CreatedAt != 0
 		got.CreatedAt = 0
@@ -241,6 +266,29 @@ func TestAnalyzer_Analyze(t *testing.T) {
 	}
 }
 
+func TestAnalyzer_recordLabels(t *testing.T) {
+	tests := []struct {
+		name       string
+		createErr  error
+		wantLogged int
+	}{
+		{name: "recorded labels are not logged"},
+		{name: "already recorded labels are not logged", createErr: storage.ErrAlreadyExists},
+		{name: "failed writes are logged", createErr: analyzerTestFailure, wantLogged: len(analyzerTestAffected)},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			h := newAnalyzerHarness(t)
+			labels := analyzerTestLabels()
+			h.dictionary.EXPECT().Create(gomock.Any(), gomock.Any()).Return(tt.createErr).Times(len(labels))
+
+			h.analyzer.(*analyzer).recordLabels(context.Background(), analyzerTestQueue, labels)
+
+			assert.Equal(t, tt.wantLogged, h.logs.Len())
+		})
+	}
+}
+
 func TestAnalyzer_Analyze_NoSignatureIsDistinctFromStoreError(t *testing.T) {
 	h := newAnalyzerHarness(t)
 	h.expectSigningNewBatch(nil)