Merge #97779

97779: kvserver: avoid reusing proposal in tryReproposeWithNewLeaseIndex r=erikgrinaker a=tbg

The previous reproposal mechanism is quite complex. We can side-step a
good amount of the complexity by not re-using proposals. Instead, when
the entry associated to a local proposal gets rejected by the
`LeaseAppliedIndex`, we mint a "new" proposal (identical to the old but
with a new command ID) which is now responsible for notifying the
caller.

This is introduced piecemeal, and the "old way" is preserved for continued
metamorphic testing (and a guaranteed way to be able to fall back, should
we see fallout over the next couple of weeks).

Phasing out the gating var `useReproposalsV2` is tracked separately[^1].

[^1]: #105625

Epic: CRDB-25287
Release note: None


Co-authored-by: Tobias Grieger <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/BUILD.bazel

@@ -297,6 +297,7 @@ go_test(
         "range_log_test.go",
         "rebalance_objective_test.go",
         "replica_application_cmd_buf_test.go",
+        "replica_application_result_test.go",
         "replica_application_state_machine_test.go",
         "replica_batch_updates_test.go",
         "replica_circuit_breaker_test.go",

pkg/kv/kvserver/replica_application_cmd.go

@@ -52,6 +52,9 @@ type replicatedCmd struct {
 	// finishTracingSpan. This span "follows from" the proposer's span (even
 	// when the proposer is remote; we marshall tracing info through the
 	// proposal).
+	//
+	// For local commands, we also have a `ProposalData` which may have a span
+	// as well, and if it does, this span will follow from it.
 	sp *tracing.Span
 
 	// splitMergeUnlock is acquired for splits and merges when they are staged

pkg/kv/kvserver/replica_application_decoder.go

@@ -17,8 +17,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/apply"
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/logcrash"
 	"github.com/cockroachdb/cockroach/pkg/util/quotapool"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/errors"
 	"go.etcd.io/raft/v3/raftpb"
 )
 
@@ -73,10 +75,77 @@ func (d *replicaDecoder) decode(ctx context.Context, ents []raftpb.Entry) error
 	return nil
 }
 
+// retrieveLocalProposalsV2 is used with useReproposalsV2, replacing a call
+// to retrieveLocalProposals. The V2 implementation is simpler because a log
+// entry that comes up for a local proposal can always consume that proposal
+// from the map because V2 never mutates the MaxLeaseIndex for the same proposal.
+// In contrast, with V1, we can only remove the proposal from the map once we
+// have found a log entry that had a matching MaxLeaseIndex. This lead to the
+// complexity of having multiple entries associated to the same proposal during
+// application.
+func (d *replicaDecoder) retrieveLocalProposalsV2() (anyLocal bool) {
+	d.r.mu.Lock()
+	defer d.r.mu.Unlock()
+
+	var it replicatedCmdBufSlice
+
+	for it.init(&d.cmdBuf); it.Valid(); it.Next() {
+		cmd := it.cur()
+		cmd.proposal = d.r.mu.proposals[cmd.ID]
+		var alloc *quotapool.IntAlloc
+		if cmd.proposal != nil {
+			// INVARIANT: a proposal is consumed (i.e. removed from the proposals map)
+			// the first time it comes up for application. (If the proposal fails due
+			// to an illegal LeaseAppliedIndex, a new proposal might be spawned to
+			// retry but that proposal will be unrelated as far as log application is
+			// concerned).
+			//
+			// INVARIANT: local proposals are not in the proposals map while being
+			// applied, and they never re-enter the proposals map either during or
+			// afterwards.
+			//
+			// (propBuf.{Insert,ReinsertLocked} ignores proposals that have
+			// v2SeenDuringApplicationSet to make this true).
+			if cmd.proposal.v2SeenDuringApplication {
+				err := errors.AssertionFailedf("ProposalData seen twice during application: %+v", cmd.proposal)
+				logcrash.ReportOrPanic(d.r.AnnotateCtx(cmd.ctx), &d.r.store.ClusterSettings().SV, "%v", err)
+				// If we didn't panic, treat the proposal as non-local. This makes sure
+				// we don't repropose it under a new lease index.
+				cmd.proposal = nil
+			} else {
+				cmd.proposal.v2SeenDuringApplication = true
+				anyLocal = true
+				delete(d.r.mu.proposals, cmd.ID)
+				if d.r.mu.proposalQuota != nil {
+					alloc = cmd.proposal.quotaAlloc
+					cmd.proposal.quotaAlloc = nil
+				}
+			}
+		}
+
+		// NB: this may append nil. It's intentional. The quota release queue
+		// needs to have one slice entry per entry applied (even if the entry
+		// is rejected).
+		//
+		// TODO(tbg): there used to be an optimization where we'd elide mutating
+		// this slice until we saw a local proposal under a populated
+		// b.r.mu.proposalQuota. We can bring it back.
+		if d.r.mu.proposalQuota != nil {
+			d.r.mu.quotaReleaseQueue = append(d.r.mu.quotaReleaseQueue, alloc)
+		}
+	}
+	return anyLocal
+}
+
 // retrieveLocalProposals binds each of the decoder's commands to their local
 // proposals if they were proposed locally. The method also sets the ctx fields
 // on all commands.
 func (d *replicaDecoder) retrieveLocalProposals(ctx context.Context) (anyLocal bool) {
+	if useReproposalsV2 {
+		// NB: we *must* use this new code for correctness, since we have an invariant
+		// described within.
+		return d.retrieveLocalProposalsV2()
+	}
 	d.r.mu.Lock()
 	defer d.r.mu.Unlock()
 	// Assign all the local proposals first then delete all of them from the map

pkg/kv/kvserver/replica_application_result.go

@@ -16,16 +16,27 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/raftlog"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary/rspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
+	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/errors"
 	"go.etcd.io/raft/v3"
 )
 
+// useReproposalsV2 activates prototype code that instead of reproposing using a
+// modified lease index makes a new proposal (different CmdID), transfers the
+// waiting caller (if any) to it, and proposes that. With this strategy, the
+// *RaftCommand associated to a proposal becomes immutable, which simplifies the
+// mental model and allows various simplifications in the proposal pipeline. For
+// now, the old and new behavior coexist, and we want to keep exercising both.
+// Once we have confidence, we'll hard-code true and remove all old code paths.
+var useReproposalsV2 = util.ConstantWithMetamorphicTestBool("reproposals-v2", true)
+
 // replica_application_*.go files provide concrete implementations of
 // the interfaces defined in the storage/apply package:
 //
@@ -105,6 +116,16 @@ func (r *Replica) prepareLocalResult(ctx context.Context, cmd *replicatedCmd) {
 		case kvserverbase.ProposalRejectionPermanent:
 			cmd.response.Err = pErr
 		case kvserverbase.ProposalRejectionIllegalLeaseIndex:
+			if useReproposalsV2 {
+				// If we're using V2 reproposals, this proposal is actually going to
+				// be fully rejected, but the client won't be listening to it at that
+				// point any more. But we should set the error. (This ends up being
+				// inconsequential but it's the right thing to do).
+				//
+				// TODO(tbg): once useReproposalsV2 is baked in, set the error unconditionally
+				// above the `switch`.
+				cmd.response.Err = pErr
+			}
 			// Reset the error as it's now going to be determined by the outcome of
 			// reproposing (or not); note that tryReproposeWithNewLeaseIndex will
 			// return `nil` if the entry is not eligible for reproposals.
@@ -171,7 +192,18 @@ func (r *Replica) prepareLocalResult(ctx context.Context, cmd *replicatedCmd) {
 				}
 			}
 			if pErr == nil { // since we might have injected an error
-				pErr = r.tryReproposeWithNewLeaseIndex(ctx, cmd)
+				if useReproposalsV2 {
+					pErr = kvpb.NewError(r.tryReproposeWithNewLeaseIndexV2(ctx, cmd))
+					if pErr == nil {
+						// Avoid falling through below. We managed to repropose, but this
+						// proposal is still erroring out. We don't want to assign to
+						// localResult. If there is an error though, we do fall through into
+						// the existing tangle of correct but unreadable handling below.
+						return
+					}
+				} else {
+					pErr = r.tryReproposeWithNewLeaseIndex(ctx, cmd)
+				}
 			}
 
 			if pErr != nil {
@@ -210,10 +242,16 @@ func (r *Replica) prepareLocalResult(ctx context.Context, cmd *replicatedCmd) {
 				// https://github.com/cockroachdb/cockroach/issues/97633
 				log.Infof(ctx, "failed to repropose %s at idx %d with new lease index: %s", cmd.ID, cmd.Index(), pErr)
 				cmd.response.Err = pErr
-			} else {
+				// Fall through.
+			} else if !useReproposalsV2 {
 				// Unbind the entry's local proposal because we just succeeded
 				// in reproposing it and we don't want to acknowledge the client
 				// yet.
+				//
+				// NB: in v2, reproposing already moved the waiting caller over to a new
+				// proposal, and by design we don't change the "Localness" of the old
+				// proposal mid-application but instead let it fail as a local proposal
+				// (which signals into an throwaway channel).
 				cmd.proposal = nil
 				return
 			}
@@ -225,6 +263,14 @@ func (r *Replica) prepareLocalResult(ctx context.Context, cmd *replicatedCmd) {
 	} else {
 		log.Fatalf(ctx, "proposal must return either a reply or an error: %+v", cmd.proposal)
 	}
+
+	// The current proposal has no error (and wasn't reproposed successfully or we
+	// would've early returned already) OR it has an error AND we failed to
+	// repropose it.
+	//
+	// TODO(tbg): it doesn't make sense to assign to `cmd.response` unconditionally.
+	// We're returning an error; the response should be nil. The error tracking in
+	// this method should be cleaned up.
 	cmd.response.EncounteredIntents = cmd.proposal.Local.DetachEncounteredIntents()
 	cmd.response.EndTxns = cmd.proposal.Local.DetachEndTxns(pErr != nil)
 	if pErr == nil {
@@ -234,6 +280,134 @@ func (r *Replica) prepareLocalResult(ctx context.Context, cmd *replicatedCmd) {
 	}
 }
 
+func (r *Replica) tryReproposeWithNewLeaseIndexV2(
+	ctx context.Context, origCmd *replicatedCmd,
+) error {
+	// NB: `origCmd` remains "Local". It's just not going to signal anyone
+	// or release any latches.
+
+	origP := origCmd.proposal
+
+	// We want to move a few items from origCmd to the new command, but only if we
+	// managed to propose the new command. For example, if we move the latches
+	// over too early but then fail to actually get the new proposal started, the
+	// old proposal will not release the latches. This would result in a lost
+	// latch.
+	var success bool
+
+	// Go through the original proposal field by field and decide what transfers
+	// to the new proposal (and how that affects the old proposal). The overall
+	// goal is that the old proposal remains a local proposal (switching it to
+	// non-local now invites logic bugs) but not bound to the caller.
+
+	// NB: quotaAlloc is always nil here, because we already
+	// released the quota unconditionally in retrieveLocalProposalsV2.
+	// So the below is a no-op.
+	//
+	// TODO(tbg): if we shifted the release of proposal quota to *after*
+	// successful application, we could move the quota over
+	// prematurely releasing it here.
+	newQuotaAlloc := origP.quotaAlloc
+	defer func() {
+		if success {
+			origP.quotaAlloc = nil
+		}
+	}()
+
+	newCommand := kvserverpb.RaftCommand{
+		ProposerLeaseSequence:   origP.command.ProposerLeaseSequence,
+		DeprecatedProposerLease: origP.command.DeprecatedProposerLease,
+		ReplicatedEvalResult:    origP.command.ReplicatedEvalResult,
+		WriteBatch:              origP.command.WriteBatch,
+		LogicalOpLog:            origP.command.LogicalOpLog,
+		TraceData:               origP.command.TraceData,
+
+		MaxLeaseIndex:       0,   // assigned on flush
+		ClosedTimestamp:     nil, // assigned on flush
+		AdmissionPriority:   0,   // assigned on flush
+		AdmissionCreateTime: 0,   // assigned on flush
+		AdmissionOriginNode: 0,   // assigned on flush
+	}
+
+	// Now we construct the remainder of the ProposalData. First, the pieces
+	// that actively "move over", i.e. those that have to do with the latches
+	// held and the caller waiting to be signaled.
+
+	// `ec` (latches, etc) transfers to the new proposal.
+	newEC := origP.ec
+	defer func() {
+		if success {
+			origP.ec = endCmds{}
+		}
+	}()
+
+	// Ditto doneCh (signal to proposer).
+	newDoneCh := origP.doneCh
+	defer func() {
+		if success {
+			origP.doneCh = nil
+		}
+	}()
+
+	r.mu.RLock()
+	ticks := r.mu.ticks
+	r.mu.RUnlock()
+
+	// TODO(tbg): work on the lifecycle of ProposalData. This struct (and the
+	// surrounding replicatedCmd) are populated in an overly ad-hoc manner.
+	// TODO(tbg): the fields are spelled out here to make explicit what is being copied
+	// here. Add a unit test that fails on addition of a new field and points at the
+	// need to double check what the intended behavior of the new field in this method
+	// is.
+	newProposal := &ProposalData{
+		// The proposal's context and span carry over. Recall that they are *not*
+		// used for command application; `cmd.{ctx,sp}` are; and since this last
+		// span "follows from" the proposal's span, if the proposal sticks around
+		// for (some reincarnation of) the command to eventually apply, its trace
+		// will reflect the reproposal as well.
+		ctx:             origP.ctx,
+		sp:              origP.sp, // NB: special handling below
+		idKey:           raftlog.MakeCmdIDKey(),
+		proposedAtTicks: 0, // set in registerProposalLocked
+		createdAtTicks:  ticks,
+		command:         &newCommand,
+		quotaAlloc:      newQuotaAlloc,
+		ec:              newEC,
+		applied:         false,
+		doneCh:          newDoneCh,
+		// Local is copied over. It won't be used on the old proposal (since that
+		// proposal got rejected), but since it's still "local" we don't want to put
+		// it into  an undefined state by removing its response. The same goes for
+		// Request.
+		Local:                   origP.Local,
+		Request:                 origP.Request,
+		leaseStatus:             origP.leaseStatus,
+		tok:                     TrackedRequestToken{}, // filled in in `propose`
+		encodedCommand:          nil,
+		raftAdmissionMeta:       nil,
+		v2SeenDuringApplication: false,
+	}
+	// If the original proposal had an explicit span, it's an async consensus
+	// proposal and the span would be finished momentarily (when we return to
+	// the caller) if we didn't unlink it here, but we want it to continue
+	// tracking newProposal. We leave it in `origP.ctx` though, since that
+	// context will become unused once the application of this (soft-failed)
+	// proposal concludes, i.e. soon after this method returns, in case there
+	// is anything left to log into it.
+	defer func() {
+		if success {
+			origP.sp = nil
+		}
+	}()
+
+	if err := r.tryReproposeWithNewLeaseIndexShared(ctx, newProposal).GoError(); err != nil {
+		return err
+	}
+
+	success = true
+	return nil
+}
+
 // tryReproposeWithNewLeaseIndex is used by prepareLocalResult to repropose
 // commands that have gotten an illegal lease index error, and that we know
 // could not have applied while their lease index was valid (that is, we
@@ -267,7 +441,12 @@ func (r *Replica) tryReproposeWithNewLeaseIndex(
 		// succeeding in the Raft log for a given command.
 		return nil
 	}
+	return r.tryReproposeWithNewLeaseIndexShared(ctx, cmd.proposal)
+}
 
+func (r *Replica) tryReproposeWithNewLeaseIndexShared(
+	ctx context.Context, p *ProposalData,
+) *kvpb.Error {
 	// We need to track the request again in order to protect its timestamp until
 	// it gets reproposed.
 	// TODO(andrei): Only track if the request consults the ts cache. Some
@@ -299,7 +478,7 @@ func (r *Replica) tryReproposeWithNewLeaseIndex(
 	if pErr != nil {
 		return pErr
 	}
-	log.VEventf(ctx, 2, "reproposed command %x", cmd.ID)
+	log.VEventf(ctx, 2, "reproposed command %x", p.idKey)
 	return nil
 }