Merge #98268 #98489

98268: kvcoord: Close done context if init fails r=miretskiy a=miretskiy

When we close mux rangefeed init context due to an error, it is important to also close client done context so that this error can be propagated to the caller.

Epic: None

Release note: None

98489: upgradejob: initialize the kvdb on TenantDeps r=JeffSwenson a=JeffSwenson

The KVDB field was created as part of PR #93218, but it was never initialized.

Part of #94843

Release Note: None

Co-authored-by: Yevgeniy Miretskiy <[email protected]>
Co-authored-by: Jeff <[email protected]>

pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go

@@ -248,8 +248,9 @@ func (m *rangefeedMuxer) startNodeMuxRangeFeed(
 	ms.initCtx.close(err)
 
 	if err == nil {
-		doneCtx.close(m.receiveEventsFromNode(ctx, ms))
+		err = m.receiveEventsFromNode(ctx, ms)
 	}
+	doneCtx.close(err)
 
 	// We propagated error to the caller via init/done context.
 	return nil //nolint:returnerrcheck
@@ -289,9 +290,13 @@ func (m *rangefeedMuxer) demuxLoop(ctx context.Context) (retErr error) {
 			case <-ctx.Done():
 				return ctx.Err()
 			case producer.eventCh <- &e.RangeFeedEvent:
+			case <-producer.callerCtx.Done():
+				if log.V(1) {
+					log.Infof(ctx, "received stray event, but caller exited: stream=%d: e=%v", e.StreamID, e)
+				}
 			case <-producer.muxClientCtx.Done():
 				if log.V(1) {
-					log.Infof(ctx, "received stray event stream %d: %v", e.StreamID, e)
+					log.Infof(ctx, "received stray event, but node mux exited: stream=%d: e=%v", e.StreamID, e)
 				}
 			}
 		}

pkg/upgrade/upgradejob/upgrade_job.go

@@ -91,6 +91,7 @@ func (r resumer) Resume(ctx context.Context, execCtxI interface{}) error {
 			Codec:            execCtx.ExecCfg().Codec,
 			Settings:         execCtx.ExecCfg().Settings,
 			DB:               execCtx.ExecCfg().InternalDB,
+			KVDB:             execCtx.ExecCfg().DB,
 			LeaseManager:     execCtx.ExecCfg().LeaseManager,
 			InternalExecutor: ex,
 			JobRegistry:      execCtx.ExecCfg().JobRegistry,