kv: address review follow-ups

Author: bootjp

adapter/dynamodb.go

@@ -62,10 +62,10 @@ const (
 	batchWriteItemMaxItems      = 25
 	dynamoMaxRequestBodyBytes   = 1 << 20
 
-	dynamoTableMetaPrefix       = "!ddb|meta|table|"
-	dynamoTableGenerationPrefix = "!ddb|meta|gen|"
-	dynamoItemPrefix            = "!ddb|item|"
-	dynamoGSIPrefix             = "!ddb|gsi|"
+	dynamoTableMetaPrefix       = kv.DynamoTableMetaPrefix
+	dynamoTableGenerationPrefix = kv.DynamoTableGenerationPrefix
+	dynamoItemPrefix            = kv.DynamoItemPrefix
+	dynamoGSIPrefix             = kv.DynamoGSIPrefix
 	dynamoScanPageLimit         = 1024
 	dynamoKeyEscapeByte         = byte(0x00)
 	dynamoKeyTerminatorByte     = byte(0x01)

kv/shard_key.go

@@ -12,19 +12,24 @@ const redisInternalRoutePrefix = "!redis|"
 var redisInternalRoutePrefixBytes = []byte(redisInternalRoutePrefix)
 
 const (
-	dynamoRoutePrefix           = "!ddb|route|table|"
-	dynamoTableMetaPrefix       = "!ddb|meta|table|"
-	dynamoTableGenerationPrefix = "!ddb|meta|gen|"
-	dynamoItemPrefix            = "!ddb|item|"
-	dynamoGSIPrefix             = "!ddb|gsi|"
+	dynamoRoutePrefix = "!ddb|route|table|"
+
+	// DynamoTableMetaPrefix prefixes DynamoDB table metadata keys.
+	DynamoTableMetaPrefix = "!ddb|meta|table|"
+	// DynamoTableGenerationPrefix prefixes DynamoDB table generation keys.
+	DynamoTableGenerationPrefix = "!ddb|meta|gen|"
+	// DynamoItemPrefix prefixes DynamoDB item storage keys.
+	DynamoItemPrefix = "!ddb|item|"
+	// DynamoGSIPrefix prefixes DynamoDB GSI storage keys.
+	DynamoGSIPrefix = "!ddb|gsi|"
 )
 
 var (
 	dynamoRoutePrefixBytes           = []byte(dynamoRoutePrefix)
-	dynamoTableMetaPrefixBytes       = []byte(dynamoTableMetaPrefix)
-	dynamoTableGenerationPrefixBytes = []byte(dynamoTableGenerationPrefix)
-	dynamoItemPrefixBytes            = []byte(dynamoItemPrefix)
-	dynamoGSIPrefixBytes             = []byte(dynamoGSIPrefix)
+	dynamoTableMetaPrefixBytes       = []byte(DynamoTableMetaPrefix)
+	dynamoTableGenerationPrefixBytes = []byte(DynamoTableGenerationPrefix)
+	dynamoItemPrefixBytes            = []byte(DynamoItemPrefix)
+	dynamoGSIPrefixBytes             = []byte(DynamoGSIPrefix)
 )
 
 // routeKey normalizes internal keys (e.g., list metadata/items) to the logical

kv/shard_key_test.go

@@ -36,10 +36,10 @@ func TestRouteKey_NormalizesDynamoKeysToTable(t *testing.T) {
 	indexSegment := base64.RawURLEncoding.EncodeToString([]byte("status-index"))
 	want := dynamoRouteTableKey(tableSegment)
 
-	metaKey := append([]byte(dynamoTableMetaPrefix), tableSegment...)
-	generationKey := append([]byte(dynamoTableGenerationPrefix), tableSegment...)
-	itemKey := append([]byte(dynamoItemPrefix+string(tableSegment)+"|7|"), []byte("pk\x00\x01")...)
-	gsiKey := append([]byte(dynamoGSIPrefix+string(tableSegment)+"|7|"+indexSegment+"|"), []byte("idx\x00\x01pk\x00\x01")...)
+	metaKey := append([]byte(DynamoTableMetaPrefix), tableSegment...)
+	generationKey := append([]byte(DynamoTableGenerationPrefix), tableSegment...)
+	itemKey := append([]byte(DynamoItemPrefix+string(tableSegment)+"|7|"), []byte("pk\x00\x01")...)
+	gsiKey := append([]byte(DynamoGSIPrefix+string(tableSegment)+"|7|"+indexSegment+"|"), []byte("idx\x00\x01pk\x00\x01")...)
 
 	require.Equal(t, want, routeKey(metaKey))
 	require.Equal(t, want, routeKey(generationKey))

kv/sharded_coordinator.go

@@ -516,16 +516,9 @@ func (c *ShardedCoordinator) txnLogs(reqs *OperationGroup[OP]) ([]*pb.Request, e
 	if len(gids) != 1 {
 		return nil, errors.WithStack(ErrInvalidRequest)
 	}
-	commitTS := reqs.CommitTS
-	if commitTS == 0 {
-		commitTS = c.nextTxnTSAfter(reqs.StartTS)
-	} else {
-		// Observe caller-provided commitTS to keep the HLC monotonic; without
-		// this the clock could later issue timestamps smaller than commitTS.
-		c.clock.Observe(commitTS)
-	}
-	if commitTS == 0 || commitTS <= reqs.StartTS {
-		return nil, errors.WithStack(ErrTxnCommitTSRequired)
+	commitTS, err := c.resolveTxnCommitTS(reqs.StartTS, reqs.CommitTS)
+	if err != nil {
+		return nil, err
 	}
 	return buildTxnLogs(reqs.StartTS, commitTS, grouped, gids)
 }

kv/txn_codec.go

@@ -25,6 +25,7 @@ const txnLockFlagPrimary byte = 0x01
 const (
 	txnMetaFlagLockTTL  byte = 0x01
 	txnMetaFlagCommitTS byte = 0x02
+	txnMetaKnownFlags   byte = txnMetaFlagLockTTL | txnMetaFlagCommitTS
 )
 
 const txnMetaHeaderSize = 2
@@ -139,6 +140,9 @@ func decodeTxnMetaV2(b []byte) (TxnMeta, error) {
 		return TxnMeta{}, errors.New("txn meta: truncated flags")
 	}
 	flags := b[1]
+	if flags&^txnMetaKnownFlags != 0 {
+		return TxnMeta{}, errors.WithStack(errors.Newf("txn meta: unsupported flags 0x%02x", flags))
+	}
 	r := bytes.NewReader(b[txnMetaHeaderSize:])
 	primaryLen, err := readTxnUint64(r, "txn meta: primary key length truncated")
 	if err != nil {
@@ -162,6 +166,9 @@ func decodeTxnMetaV2(b []byte) (TxnMeta, error) {
 			return TxnMeta{}, err
 		}
 	}
+	if r.Len() != 0 {
+		return TxnMeta{}, errors.WithStack(errors.Newf("txn meta: unexpected trailing bytes %d", r.Len()))
+	}
 	return meta, nil
 }
 

kv/txn_codec_test.go

@@ -74,3 +74,22 @@ func TestDecodeTxnMetaV1Compatibility(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, meta, decoded)
 }
+
+func TestDecodeTxnMetaV2RejectsUnknownFlags(t *testing.T) {
+	t.Parallel()
+
+	encoded := encodeTxnMetaV2(TxnMeta{PrimaryKey: []byte("pk")})
+	encoded[1] |= 0x80
+
+	_, err := DecodeTxnMeta(encoded)
+	require.ErrorContains(t, err, "unsupported flags")
+}
+
+func TestDecodeTxnMetaV2RejectsTrailingBytes(t *testing.T) {
+	t.Parallel()
+
+	encoded := append(encodeTxnMetaV2(TxnMeta{PrimaryKey: []byte("pk")}), 0x01)
+
+	_, err := DecodeTxnMeta(encoded)
+	require.ErrorContains(t, err, "unexpected trailing bytes")
+}