raft: enable accepted-baseline protected startup

adapter/distribution_server.go

@@ -1544,25 +1544,9 @@ func nextCatalogSplitIDs(expectedVersion uint64, rightRouteID uint64) (uint64, u
 }
 
 func catalogStoreMutationsToOps(mutations []*store.KVPairMutation) ([]*kv.Elem[kv.OP], error) {
-	ops := make([]*kv.Elem[kv.OP], 0, len(mutations))
-	for _, mutation := range mutations {
-		if mutation == nil {
-			return nil, errors.WithStack(errDistributionCatalogMutationInvalid)
-		}
-		var op kv.OP
-		switch mutation.Op {
-		case store.OpTypePut:
-			op = kv.Put
-		case store.OpTypeDelete:
-			op = kv.Del
-		default:
-			return nil, errors.Wrapf(errDistributionCatalogMutationInvalid, "unknown operation %d", mutation.Op)
-		}
-		ops = append(ops, &kv.Elem[kv.OP]{
-			Op:    op,
-			Key:   distribution.CloneBytes(mutation.Key),
-			Value: distribution.CloneBytes(mutation.Value),
-		})
+	ops, err := kv.StoreMutationsToElems(mutations)
+	if err != nil {
+		return nil, errors.WithStack(errors.Join(errDistributionCatalogMutationInvalid, err))
 	}
 	return ops, nil
 }

adapter/dynamodb.go

@@ -118,17 +118,18 @@ const (
 type DynamoDBServerOption func(*DynamoDBServer)
 
 type DynamoDBServer struct {
-	listen          net.Listener
-	store           store.MVCCStore
-	coordinator     kv.Coordinator
-	httpServer      *http.Server
-	httpConns       *httpConnTracker
-	targetHandlers  map[string]func(http.ResponseWriter, *http.Request)
-	readTracker     *kv.ActiveTimestampTracker
-	requestObserver monitoring.DynamoDBRequestObserver
-	itemUpdateLocks [itemUpdateLockStripeCount]chan struct{}
-	tableLocks      [tableLockStripeCount]sync.Mutex
-	leaderDynamo    map[string]string
+	requestAdmission RequestAdmission
+	listen           net.Listener
+	store            store.MVCCStore
+	coordinator      kv.Coordinator
+	httpServer       *http.Server
+	httpConns        *httpConnTracker
+	targetHandlers   map[string]func(http.ResponseWriter, *http.Request)
+	readTracker      *kv.ActiveTimestampTracker
+	requestObserver  monitoring.DynamoDBRequestObserver
+	itemUpdateLocks  [itemUpdateLockStripeCount]chan struct{}
+	tableLocks       [tableLockStripeCount]sync.Mutex
+	leaderDynamo     map[string]string
 
 	// onePhaseTxnDedup enables option-2 one-phase idempotency on the
 	// single-item write path (UpdateItem / PutItem / DeleteItem): on a
@@ -302,6 +303,7 @@ func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Co
 			opt(d)
 		}
 	}
+	installHTTPRequestAdmission(d.httpServer, d.requestAdmission)
 	return d
 }
 

adapter/internal.go

@@ -177,6 +177,8 @@ type Internal struct {
 	clock                           *kv.HLC
 	tsAllocator                     kv.TimestampAllocator
 	adminProposer                   raftengine.Proposer
+	bootstrapPolicyProposer         raftengine.Proposer
+	prepareStartupAuthorization     func(context.Context, *pb.PrepareEvidenceStateRequest) bool
 	readWatermarkProposer           raftengine.Proposer
 	lastCommitTimestamp             func() uint64
 	relay                           *RedisPubSubRelay
@@ -605,8 +607,18 @@ func (i *Internal) ForwardAdminProposal(
 		return nil, errors.WithStack(ErrNotLeader)
 	}
 	if ts, ok, err := kv.DecodeWatermarkAdvance(req.GetPayload()); ok {
-		return i.forwardReadWatermarkAdvance(ctx, req.GetPayload(), ts, err)
+		return i.forwardReadWatermarkAdvance(ctx, req.GetPayload(), ts, req.GetForceEntry(), err)
 	}
+	if req.GetForceEntry() {
+		return nil, errors.WithStack(status.Error(codes.PermissionDenied, "force-entry forwarding is only valid for read watermark advances"))
+	}
+	if reply, matched, err := i.forwardBootstrapActivation(ctx, req); matched {
+		return reply, err
+	}
+	return i.forwardSupportedAdminProposal(ctx, req)
+}
+
+func (i *Internal) forwardSupportedAdminProposal(ctx context.Context, req *pb.ForwardAdminProposalRequest) (*pb.ForwardAdminProposalResponse, error) {
 	if i.adminProposer == nil {
 		return nil, errors.New("admin proposer is unavailable")
 	}
@@ -624,19 +636,20 @@ func (i *Internal) ForwardAdminProposal(
 }
 
 // forwardReadWatermarkAdvance applies a read-watermark advance a follower
-// forwarded before serving a snapshot read on this group. It is a no-op when
-// the group already covers ts, and refuses a timestamp beyond the HLC bound
-// so a caller cannot fence the group's writes into the future.
+// forwarded before serving a snapshot read on this group. Ordinary reads are
+// a no-op when the group already covers ts; explicit receipt requests force a
+// new entry. Both modes refuse timestamps beyond the HLC bound.
 func (i *Internal) forwardReadWatermarkAdvance(
 	ctx context.Context,
 	payload []byte,
 	ts uint64,
+	forceEntry bool,
 	decodeErr error,
 ) (*pb.ForwardAdminProposalResponse, error) {
 	if decodeErr != nil {
 		return nil, errors.WithStack(status.Error(codes.PermissionDenied, "admin forwarding requires a supported typed admin entry"))
 	}
-	if i.store != nil && ts <= i.store.LastCommitTS() {
+	if !forceEntry && i.store != nil && ts <= i.store.LastCommitTS() {
 		return &pb.ForwardAdminProposalResponse{}, nil
 	}
 	if i.readWatermarkProposer == nil {

adapter/internal_bootstrap_policy.go

@@ -0,0 +1,65 @@
+package adapter
+
+import (
+	"context"
+
+	"github.com/bootjp/elastickv/internal/peersession"
+	"github.com/bootjp/elastickv/internal/raftbootstrap"
+	"github.com/bootjp/elastickv/internal/raftengine"
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/cockroachdb/errors"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// This proposer is separate from the general Admin surface. The shared listener
+// must independently enforce its typed mTLS session, startup token and A handle.
+func WithInternalBootstrapPolicyProposer(proposer raftengine.Proposer) InternalOption {
+	return func(server *Internal) { server.bootstrapPolicyProposer = proposer }
+}
+
+// The startup owner must verify the typed peer receipt, dedicated credential
+// and current private admission handle. A header or context boolean is not proof.
+func WithInternalPrepareEvidenceStartupAuthorization(authorize func(context.Context, *pb.PrepareEvidenceStateRequest) bool) InternalOption {
+	return func(server *Internal) { server.prepareStartupAuthorization = authorize }
+}
+
+func (i *Internal) authorizePrepareEvidenceRequest(ctx context.Context, req *pb.PrepareEvidenceStateRequest) error {
+	if i.prepareStartupAuthorization != nil && i.prepareStartupAuthorization(ctx, req) {
+		return nil
+	}
+	return checkBearerToken(ctx, i.prepareEvidenceToken, "prepare evidence peer")
+}
+
+func (i *Internal) forwardBootstrapActivation(ctx context.Context, req *pb.ForwardAdminProposalRequest) (*pb.ForwardAdminProposalResponse, bool, error) {
+	command, matched, err := raftbootstrap.ParseActivationCommand(req.GetPayload())
+	if !matched {
+		return nil, false, nil
+	}
+	if err != nil || req.GetForceEntry() || i.bootstrapPolicyProposer == nil {
+		return nil, true, errors.WithStack(status.Error(codes.PermissionDenied, "bootstrap activation unavailable or malformed"))
+	}
+	receipt := peersession.CapturedReceipt(ctx)
+	if !bootstrapActivationReceiptMatches(receipt, command) {
+		return nil, true, errors.WithStack(status.Error(codes.PermissionDenied, "typed bootstrap policy session required"))
+	}
+	if err := receipt.Check(ctx); err != nil {
+		return nil, true, errors.WithStack(err)
+	}
+	result, err := i.bootstrapPolicyProposer.ProposeAdmin(ctx, req.GetPayload())
+	if err != nil {
+		return nil, true, errors.WithStack(err)
+	}
+	if err := forwardedAdminProposalResponseError(result); err != nil {
+		return nil, true, err
+	}
+	return &pb.ForwardAdminProposalResponse{CommitIndex: result.CommitIndex}, true, nil
+}
+
+func bootstrapActivationReceiptMatches(receipt *peersession.Receipt, command raftbootstrap.ActivationCommand) bool {
+	if receipt == nil {
+		return false
+	}
+	message := receipt.Message()
+	return message.Traffic == peersession.PolicyControl && message.Group == command.Witness.Group
+}

adapter/internal_prepare_evidence.go

@@ -24,7 +24,7 @@ func WithInternalPrepareEvidenceToken(token string) InternalOption {
 // GetPrepareEvidenceState serves read-only protocol negotiation on every
 // replica, and authoritative drain inspection only on an applied leader.
 func (i *Internal) GetPrepareEvidenceState(ctx context.Context, req *pb.PrepareEvidenceStateRequest) (*pb.PrepareEvidenceStateResponse, error) {
-	if err := checkBearerToken(ctx, i.prepareEvidenceToken, "prepare evidence peer"); err != nil {
+	if err := i.authorizePrepareEvidenceRequest(ctx, req); err != nil {
 		return nil, err
 	}
 	if req == nil {

adapter/internal_read_watermark_test.go

@@ -76,6 +76,31 @@ func TestInternalForwardReadWatermarkAdvanceCoveredIsNoop(t *testing.T) {
 	require.Nil(t, proposer.payload)
 }
 
+func TestInternalForwardExplicitReadWatermarkAdvanceCoveredStillProposes(t *testing.T) {
+	t.Parallel()
+	st := store.NewMVCCStore()
+	require.NoError(t, st.ApplyMutationsRaftAt(context.Background(), nil, nil, 50, 50, 1))
+	proposer := &internalAdminProposer{}
+	internal, _ := newReadWatermarkInternal(t, proposer, WithInternalStore(st))
+	payload := kv.EncodeWatermarkAdvance(50)
+
+	resp, err := internal.ForwardAdminProposal(context.Background(), &pb.ForwardAdminProposalRequest{Payload: payload, ForceEntry: true})
+	require.NoError(t, err)
+	require.Equal(t, uint64(17), resp.GetCommitIndex())
+	require.Equal(t, payload, proposer.payload, "an explicit fence must return a committed-entry receipt even when already covered")
+}
+
+func TestInternalForwardForceEntryRejectsNonWatermarkPayload(t *testing.T) {
+	t.Parallel()
+	proposer := &internalAdminProposer{}
+	internal, _ := newReadWatermarkInternal(t, proposer)
+	_, err := internal.ForwardAdminProposal(context.Background(), &pb.ForwardAdminProposalRequest{
+		Payload: []byte("admin"), ForceEntry: true,
+	})
+	require.Equal(t, codes.PermissionDenied, status.Code(err))
+	require.Nil(t, proposer.payload)
+}
+
 // Without the read-watermark proposer (the advance is opt-in until every
 // replica understands it) a forwarded advance is refused, and never falls
 // back to the admin proposer.

adapter/migration_ownership_refresh_test.go

@@ -14,7 +14,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
@@ -486,7 +485,7 @@ func TestOwnershipAttestationRPCRequiresMigrationCredential(t *testing.T) {
 		unary, _ := SplitMigrationTokenAuth(t.Name(), false, enforced)
 		for _, token := range []string{"", "other", t.Name()} {
 			ctx := metadata.NewIncomingContext(t.Context(), metadata.Pairs("authorization", "Bearer "+token))
-			ctx = peer.NewContext(ctx, &peer.Peer{AuthInfo: credentials.TLSInfo{}})
+			ctx = peer.NewContext(ctx, &peer.Peer{AuthInfo: migrationTLSInfoFixture(t)})
 			called := false
 			_, err := unary(ctx, &pb.AttestMigrationOwnershipRequest{}, &grpc.UnaryServerInfo{
 				FullMethod: pb.Distribution_AttestMigrationOwnership_FullMethodName,