Merge #105877

105877: kvserver: fix trace span use-after-finish in apply pipeline r=erikgrinaker a=tbg

Reported internally[^1]. It's possible for a proposal to be inserted into the
proposal buffer just before it is applied, and to be flushed when it is
applied. In this case, we would, in `propBuf.FlushLockedWithRaftGroup`, log to
the trace span, which is a problem since that span got finished when the
proposal got applied.

While this crash happened under `useReproposalsV2==true`, I'm unsure why this
would be a new problem with `useReproposalsV2`, and it doesn't seem to be
particularly common either[^2]. However, I found a small gap in our handling of
applied proposals, where one could actually show up in
`FlushLockedWithRaftGroup` and explain this crash.

I tightened the existing protections to drop the proposal before it reaches
this point more reliably. Now we should no longer see this crash; if we did
again, this would indicate a more serious problem in the trace span lifecycle.


[^1]: https://cockroachlabs.slack.com/archives/C0KB9Q03D/p1688058044517609
[^2]: `./dev test --stress ./pkg/ccl/changefeedccl/ --filter TestChangefeedExactlyOnceExport --cpus 6`: 2456 runs so far, 0 failures, over 5m0s


Epic: CRDB-25287
Release note: None


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

pkg/kv/kvserver/replica_proposal_buf.go

@@ -299,9 +299,13 @@ func (b *propBuf) Insert(ctx context.Context, p *ProposalData, tok TrackedReques
 // buffer back into the buffer to be reproposed at a new Raft log index. Unlike
 // Insert, it does not modify the command.
 func (b *propBuf) ReinsertLocked(ctx context.Context, p *ProposalData) error {
-	if p.v2SeenDuringApplication {
-		return nil
-	}
+	// NB: we can see proposals here that have already been applied
+	// (v2SeenDuringApplication==true). We want those to not be flushed again.
+	// However, we can also see a proposal here that is not applied yet while
+	// inserting but is applied by the time we flush. So we don't drop the
+	// proposal here, but instead in FlushLockedWithRaftGroup, to unify the two
+	// cases.
+
 	// Update the proposal buffer counter and determine which index we should
 	// insert at.
 	idx, err := b.allocateIndex(ctx, true /* wLocked */)
@@ -643,6 +647,12 @@ func (b *propBuf) maybeRejectUnsafeProposalLocked(
 		// the proposal or not.
 		return false
 	}
+	if p.v2SeenDuringApplication {
+		// Due to `refreshProposalsLocked`, we can end up with proposals that are
+		// already applied. We just want to drop those on the floor as we know
+		// that they have fully been taken care of.
+		return true
+	}
 	switch {
 	case p.Request.IsSingleRequestLeaseRequest():
 		// Handle an edge case about lease acquisitions: we don't want to forward