consumer: Apply RPC takes optional hints to set on upsert

Certain migration use-cases require that shards new to a cluster also have
pre-existing FSM hints available for their recovery log.

Piggy-back optional FSM hints onto the Apply RPC which creates the shard,
so that they're created in the same Etcd transaction.

Author: jgraettinger

consumer/protocol/protocol.pb.go

@@ -358,6 +358,11 @@ message ApplyRequest {
     ShardSpec upsert = 2;
     // Shard to be deleted. expect_mod_revision must not be zero.
     string delete = 3 [ (gogoproto.casttype) = "ShardID" ];
+    // (Optional) Primary hints to set for the shard. This is rarely required,
+    // as shards start with an empty recovery log and track their own hints,
+    // but is useful in some advanced migration scenarios.
+    // If set, then this Change must be an upsert.
+    recoverylog.FSMHints primary_hints = 4;
   }
   repeated Change changes = 1 [ (gogoproto.nullable) = false ];
   // Optional extension of the ApplyRequest.

consumer/protocol/protocol.proto

@@ -97,11 +97,19 @@ func (m *ApplyRequest_Change) Validate() error {
 		} else if m.ExpectModRevision < 0 && (m.ExpectModRevision != -1) {
 			return pb.NewValidationError("invalid ExpectModRevision (%d; expected >= 0 or -1)", m.ExpectModRevision)
 		}
+
+		if m.PrimaryHints == nil {
+			// Pass.
+		} else if err := m.PrimaryHints.Validate(); err != nil {
+			return pb.ExtendContext(err, "PrimaryHints")
+		}
 	} else if m.Delete != "" {
 		if err := m.Delete.Validate(); err != nil {
 			return pb.ExtendContext(err, "Delete")
 		} else if m.ExpectModRevision <= 0 && (m.ExpectModRevision != -1) {
 			return pb.NewValidationError("invalid ExpectModRevision (%d; expected > 0 or -1)", m.ExpectModRevision)
+		} else if m.PrimaryHints != nil {
+			return pb.NewValidationError("hints may be set only with an upsert, not a delete")
 		}
 	} else {
 		return pb.NewValidationError("neither Upsert nor Delete are set (expected exactly one)")

consumer/protocol/rpc_extensions.go

@@ -2,6 +2,7 @@ package protocol
 
 import (
 	pb "go.gazette.dev/core/broker/protocol"
+	"go.gazette.dev/core/consumer/recoverylog"
 	gc "gopkg.in/check.v1"
 )
 
@@ -106,10 +107,18 @@ func (s *RPCSuite) TestApplyRequestValidationCases(c *gc.C) {
 					MaxTxnDuration:    1,
 				},
 				Delete: "another-id",
+				PrimaryHints: &recoverylog.FSMHints{
+					Log: "valid",
+					LiveNodes: []recoverylog.FnodeSegments{{
+						Fnode:    32,
+						Segments: []recoverylog.Segment{{Author: 0, FirstSeqNo: 1, LastSeqNo: 2}},
+					}},
+				},
 			},
 			{
 				ExpectModRevision: 0,
 				Delete:            "another invalid id",
+				PrimaryHints:      &recoverylog.FSMHints{Log: "invalid"},
 			},
 			{
 				ExpectModRevision: 1,
@@ -123,10 +132,14 @@ func (s *RPCSuite) TestApplyRequestValidationCases(c *gc.C) {
 	req.Changes[0].Upsert.Id = "a-valid-id"
 	c.Check(req.Validate(), gc.ErrorMatches, `Changes\[0\]: invalid ExpectModRevision \(-2; expected >= 0 or -1\)`)
 	req.Changes[0].ExpectModRevision = 0
+	c.Check(req.Validate(), gc.ErrorMatches, `Changes\[0\].PrimaryHints.Segment: Author is zero`)
+	req.Changes[0].PrimaryHints.LiveNodes[0].Segments[0].Author = 1
 	c.Check(req.Validate(), gc.ErrorMatches, `Changes\[1\].Delete: not a valid token \(another invalid id\)`)
 	req.Changes[1].Delete = "other-valid-id"
 	c.Check(req.Validate(), gc.ErrorMatches, `Changes\[1\]: invalid ExpectModRevision \(0; expected > 0 or -1\)`)
 	req.Changes[1].ExpectModRevision = 1
+	c.Check(req.Validate(), gc.ErrorMatches, `Changes\[1\]: hints may be set only with an upsert, not a delete`)
+	req.Changes[1].PrimaryHints = nil
 	c.Check(req.Validate(), gc.ErrorMatches, `Changes\[2\]: neither Upsert nor Delete are set \(expected exactly one\)`)
 	req.Changes[2].Delete = "yet-another-valid-id"
 

consumer/protocol/rpc_extensions_test.go

@@ -2,6 +2,7 @@ package consumer
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 
@@ -126,6 +127,14 @@ func ShardApply(ctx context.Context, claims pb.Claims, srv *Service, req *pc.App
 			}
 			key = allocator.ItemKey(s.KS, change.Upsert.Id.String())
 			ops = append(ops, clientv3.OpPut(key, change.Upsert.MarshalString()))
+
+			if change.PrimaryHints != nil {
+				var val, err = json.Marshal(change.PrimaryHints)
+				if err != nil {
+					return nil, errors.WithMessage(err, "json.Marshal(hints)")
+				}
+				ops = append(ops, clientv3.OpPut(change.Upsert.HintPrimaryKey(), string(val)))
+			}
 		} else {
 			if err := authorizeShard(&claims, change.Delete); err != nil {
 				return nil, err

consumer/shard_api.go

@@ -145,11 +145,12 @@ func TestAPIApplyCases(t *testing.T) {
 	var tf, cleanup = newTestFixture(t)
 	defer cleanup()
 
+	var ctx = context.Background()
 	var specA = makeShard(shardA)
 	var specB = makeShard(shardB)
 
 	var verifyAndFetchRev = func(id pc.ShardID, expect pc.ShardSpec) int64 {
-		var resp, err = tf.service.List(context.Background(), allClaims, &pc.ListRequest{
+		var resp, err = tf.service.List(ctx, allClaims, &pc.ListRequest{
 			Selector: pb.LabelSelector{Include: pb.MustLabelSet("id", id.String())},
 		})
 		require.NoError(t, err)
@@ -158,7 +159,7 @@ func TestAPIApplyCases(t *testing.T) {
 		return resp.Shards[0].ModRevision
 	}
 	var apply = func(req *pc.ApplyRequest) *pc.ApplyResponse {
-		var resp, err = tf.service.Apply(context.Background(), allClaims, req)
+		var resp, err = tf.service.Apply(ctx, allClaims, req)
 		require.NoError(t, err)
 		return resp
 	}
@@ -167,10 +168,17 @@ func TestAPIApplyCases(t *testing.T) {
 	require.Equal(t, pc.Status_OK, apply(&pc.ApplyRequest{
 		Changes: []pc.ApplyRequest_Change{
 			{Upsert: specA},
-			{Upsert: specB},
+			{Upsert: specB, PrimaryHints: &recoverylog.FSMHints{
+				Log:        specB.RecoveryLog(),
+				Properties: []recoverylog.Property{{Path: "hello", Content: "shard"}}}},
 		},
 	}).Status)
 
+	var hints, err = tf.service.GetHints(ctx, allClaims, &pc.GetHintsRequest{Shard: shardB})
+	require.NoError(t, err)
+	require.Equal(t, recoverylog.Property{Path: "hello", Content: "shard"},
+		hints.PrimaryHints.Hints.Properties[0])
+
 	// Case: Update existing spec B.
 	var origSpecB = *specB
 	specB.Labels = append(specB.Labels, pb.Label{Name: "foo", Value: "bar"})
@@ -217,19 +225,19 @@ func TestAPIApplyCases(t *testing.T) {
 	}).Status)
 
 	// Case: Insufficient claimed selector on delete.
-	var _, err = tf.service.Apply(context.Background(), noClaims, &pc.ApplyRequest{
+	_, err = tf.service.Apply(ctx, noClaims, &pc.ApplyRequest{
 		Changes: []pc.ApplyRequest_Change{{Delete: "shard-A", ExpectModRevision: -1}},
 	})
 	require.EqualError(t, err, `rpc error: code = Unauthenticated desc = not authorized to shard-A`)
 
 	// Case: Insufficient claimed selector on upsert.
-	_, err = tf.service.Apply(context.Background(), noClaims, &pc.ApplyRequest{
+	_, err = tf.service.Apply(ctx, noClaims, &pc.ApplyRequest{
 		Changes: []pc.ApplyRequest_Change{{Upsert: specB}},
 	})
 	require.EqualError(t, err, `rpc error: code = Unauthenticated desc = not authorized to shard-B`)
 
 	// Case: Invalid requests fail with an error.
-	_, err = tf.service.Apply(context.Background(), allClaims, &pc.ApplyRequest{
+	_, err = tf.service.Apply(ctx, allClaims, &pc.ApplyRequest{
 		Changes: []pc.ApplyRequest_Change{{Delete: "invalid shard id"}},
 	})
 	require.EqualError(t, err, `Changes[0].Delete: not a valid token (invalid shard id)`)