Added a simple datadriven test framework and addressed the comments so far. This PR is missing two (small) unit tests:

• Something that overflows the underlying rangefeed buffer maintained in the KVSubscriber to ensure that we re-establish the feed correctly and always present a consistent view of a spanconfig.StoreReader
• Something at the kvserver.Store level that ensures that following a span update, we iterate through the right replicas and install a KVSubscriber.StoreReader source span config.

Will get to these tomorrow.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 1988 at r8 (raw file):

Ack, will split once ready to merge.

pkg/kv/kvserver/store.go, line 2033 at r16 (raw file):

Done.

pkg/spanconfig/spanconfig.go, line 45 at r7 (raw file):

Made this change above -- PTAL.

pkg/spanconfig/spanconfig.go, line 63 at r16 (raw file):

spanconfig.Store*; divvied up this comment block to a bunch of other places.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 134 at r16 (raw file):

Yes, in fact it's the expected pattern (see pkg/server) that KV will have provided an already started KVSubscriber to register callbacks into.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 176 at r16 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 224 at r16 (raw file):

Yes, it's a guarantee provided by the rangefeed library. Added a comment to the library itself -- see the usage of the RangeFeed.stopped channel -- it's closed by the single thread processing all the events, and .Close() on the rangefeed waits for the stopped channel to have been closed.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):

Is it possible for the frontier to advance before the initial scan is complete?

I don't think so, both from a reading of the code and intuitively from what these frontier bumps are supposed to represent -- "no updates will be observed for a timestamp less than the frontier timestamp" (how are we to guarantee that before finishing the entire scan?)

Is this because a handler can be added after the subscriber has started? If so, is being able to do so important?

Yes, it was written such that a handler could be added after the subscriber was started. We could always pull .Start into the interface itself, and rely on the caller to wire up the callback handler to the embedded StoreReader, but this felt just as easy to reason about and similar to our other uses of rangefeeds (that are started at the pkg/server level).

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 265 at r16 (raw file):

Done, both here and as part of the type-level comment.

pkg/spanconfig/spanconfigkvwatcher/span_config_decoder.go, line 50 at r8 (raw file):

BenchmarkSpanConfigDecoder-16            1911889               594.7 ns/op           242 B/op          5 allocs/op
BenchmarkSpanConfigDecoder-16            2025855               585.6 ns/op           241 B/op          5 allocs/op
BenchmarkSpanConfigDecoder-16            1982452               587.9 ns/op           242 B/op          5 allocs/op
BenchmarkSpanConfigDecoder-16            2261464               532.1 ns/op           241 B/op          5 allocs/op
BenchmarkSpanConfigDecoder-16            2220526               536.6 ns/op           241 B/op          5 allocs/op

Reviewed 1 of 17 files at r4, 2 of 105 files at r17, 18 of 29 files at r19, 10 of 10 files at r20, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 1843 at r8 (raw file):

I plan to remove it shortly once we start issuing RPCs for realsies and defaulting to using the new infrastructure; we'll need actual tests then and not just log warnings.

Do you mind leaving a TODO for this?

pkg/kv/kvserver/store.go, line 2015 at r20 (raw file):

	// We'll want to offer all replicas to the split and merge queues. Be a
	// little careful about not spawning too many individual goroutines.

What is this comment in reference to? The s.splitQueue.Async and s.mergeQueue.Async down below? How are we avoiding the 2 goroutines per range?

pkg/spanconfig/spanconfig.go, line 50 at r20 (raw file):

// state (system.span_configurations). The maintenance happens transparently;
// callers can subscribe to learn about what key spans may have seen a
// configuration change. After learning about a span update, consulting the

nit: the expected usage pattern would be clearer if this comment avoided the passive tone at the end. Consider changing the last sentence to:

// After learning about a span update through a callback invocation, callers can
// consult the embedded StoreReader to retrieve an up-to-date[2] config for the
// updated span.

pkg/spanconfig/spanconfig.go, line 69 at r20 (raw file):

//      feeds when errors occur, possibly re-transmitting earlier updates
//      (usually through a lazy [min,max) span) despite possibly not needing to.
type KVSubscriber interface {

Let's try to learn from the fact that we keep running into issues with implicit (and easily misunderstood) contracts regarding callbacks passed through interface boundaries. What can we say about this callback in the interface definition to make it harder to misuse? Which goroutine is it called on? Can multiple invocations of the callback occur concurrently?

pkg/spanconfig/spanconfigkvaccessor/kvaccessor.go, line 36 at r20 (raw file):

// CRDB cluster. It's a concrete implementation of the KVAccessor interface.
type KVAccessor struct {
	db       *kv.DB

Is the refactor in this file unrelated to the rest of the commit? If so, consider extracting it to its own commit. Same thing with the spanconfigtestutils changes.

pkg/spanconfig/spanconfigkvsubscriber/datadriven_test.go, line 142 at r20 (raw file):

					mu.Lock()
					mu.lastFrontierTS = ts
					defer mu.Unlock()

nit: either move the defer mu.Unlock() up or remove the defer.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):

I don't think so, both from a reading of the code and intuitively from what these frontier bumps are supposed to represent -- "no updates will be observed for a timestamp less than the frontier timestamp" (how are we to guarantee that before finishing the entire scan?)

Got it, thanks. Assuming we don't get our wires crossed and hear from old rangefeeds, what you say checks out.

Yes, it was written such that a handler could be added after the subscriber was started. We could always pull .Start into the interface itself, and rely on the caller to wire up the callback handler to the embedded StoreReader, but this felt just as easy to reason about and similar to our other uses of rangefeeds (that are started at the pkg/server level).

I'm surprised by this. Is this as easy to reason about? The late binding of the callback means that we have more states that the subscriber can be in. For instance, the initial scan can complete before the handler is registered. I think this is why we have to handle the handler initialization in onFrontierAdvance as well as in onInitialScanDone. It's also why we have to access the handler under the mutex

I also don't think it's similar to our other uses of rangefeeds. The rangefeed client library accepts its onValue callback on creation before its rangefeed is established.

Is there a benefit to this late binding that I'm missing?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 57 at r20 (raw file):

// initial scan, concurrent readers would continue to observe the last
// spanconfig.StoreReader if any. After the swap, it would observe the more
// up-to-date source instead. Subsequent incremental updates target the new

Can we add a claim here about why we are guaranteed that the new StoreReader will be more up-to-date than the old one by the time it is swapped out? I don't think we want to allow regressions.

Even with a claim, is there an assertion we can add somewhere to sanity check this?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 64 at r20 (raw file):

// instead of informing callers with an everything [min,max) span, we could diff
// the two data structures and only emit targeted updates.
type KVSubscriber struct {

Have you given any thought to observability of these components? Are there metrics that would help understand how they are behaving in a real cluster and diagnose poor behavior like rapid restarts?

cc. @andreimatei maybe this would be a good trial ground for some jstable-powered observability.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 86 at r20 (raw file):

		handler            func(roachpb.Span)
		handlerInitialized bool

Does handlerInitialized need to be accessed under lock? Isn't it local to the single rangefeed callback goroutine?

Combined with the comment above, it may be possible to reduce the fields under lock to just the spanconfig.StoreReader. That feels right to me after reading the comment above, as this the only state that's meant to be shared across concurrent actors.

If we do get to that state, you may consider replacing the mutex with an atomic.Value. That will make the atomic swap cleaner, it will make access cheaper (no CPU synchronization cost between concurrent readers), and it will prevent future changes from adding more state under the lock without thinking through ownership carefully.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 171 at r20 (raw file):

func (s *KVSubscriber) establishRangefeed(ctx context.Context) error {
	s.internal = spanconfigstore.New(s.fallback)

The comment Andrew and I were making yesterday was mostly about these fields and how we replace them one-by-one when restarting the rangefeed. Something about how implicit that is feels wrong. The suggestion was to wrap up all of the state scoped to a single rangefeed attempt in a struct. This would allow us to clobber it all at once on restart and have an easier time convincing ourselves (now and in the future when this code inevitably changes) that we're starting from a blank slate on each retry and aren't accidentally forgetting to clear some state and letting it leak into the next retry.

Something like:

s.cur.rangefeed.Close()
s.cur = attempt{}
s.establishRangefeed(ctx) // could even assert that s.cur is empty

If you aren't a fan of that, then it still might be a good idea to add commentary about which fields are scoped to a single retry and add a new teardownRangefeed method that Closes the rangefeed and clear all these fields.

pkg/spanconfig/spanconfigtestutils/utils.go, line 130 at r20 (raw file):

		}

		if strings.HasPrefix(line, deletePrefix) {

nit: this would avoid redundancy by being structured like:

switch {
case strings.HasPrefix(line, deletePrefix):
    ...
case strings.HasPrefix(line, upsertPrefix):
    ...
default:
    t.Fatalf("malformed line %q, expected to find prefix %q or %q",
        line, upsertPrefix, deletePrefix)
}

Added the tests above, but more importantly, I've re-worked things yet again (third time's the charm!). Instead of having the KVSubscriber being a stand-alone component that's internally retrying and re-establishing feeds, etc. I've made it so that the caller can bounce the entire subscriber all at once. Quite a lot of code was simplified as a result -- we're no longer doing the hairy "chain a bunch of rangefeed callbacks together and also bounce the rangefeed in one of these callbacks". This is very similar to what we're doing now at the SQLWatcher, #71968, which is nice. PTAL.

The top-level Subscribe call is a blocking one, returning only on errors (at which point callers would re-Subscribe). The handlers installed concurrently stay installed, and the exported StoreReader is still accessible. When successfully re-Subscribed, the handlers will learn about it and will be able to consult a fresher StoreReader for new data. This structure feels much easier because the lifecycle management of the entire Subscriber is decoupled from passing updates to the handler, and the code dealing with the subscriber lifecycle can use whatever retry.Options it wants (for now it'll retry indefinitely with some back-off).

I've broken reviewable by an accidental amend rebase, so it's kind of slow to work with. If you want a fresh slate for this PR, I'm happy to make one.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 1843 at r8 (raw file):

Done.

pkg/kv/kvserver/store.go, line 2015 at r20 (raw file):

This is in reference to the fact that we have a spanConfigUpdateQueueRateLimiter below to pace ourselves. Copy-pasta from the system config span system, though I suspect we can be much more lenient with the rate limiting here given the updates are targeted (i.e. we know exactly which ranges are changing), whereas previously it wasn't. There's a TODO to this effect near the rate limiter itself.

pkg/spanconfig/spanconfig.go, line 69 at r20 (raw file):

Single goroutine; added commentary here and near the concrete type.

pkg/spanconfig/spanconfigkvaccessor/kvaccessor.go, line 36 at r20 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/datadriven_test.go, line 142 at r20 (raw file):

Woops, done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):
I've re-worked some of this control flow to make it so that all these handlers are run in a single thread and at only two points. The overall re-working is described in the top-level comment, and I think it's a lot more "standard" usage wrt rangefeeds now. There are a lot less fields to begin with and a lot easier to reason about what's happening under the mutex. PTAL -- if it's still messy, I'm happy to fold a Start into the interface itself and disallow lazy bindings.

As for declaring the handler as "initialized" after the initial scan, it wasn't necessary -- we can wait for the subsequent frontier advance to make it so. I just did it this way to opportunistically avoid having the callers deal with back-to-back "everything" updates.

Is there a benefit to this late binding that I'm missing?

The benefit is that we can avoid having to pass in the concrete type into pkg/kv/kvserver by having pkg/server.Start oversee the lifecycle of the Subscriber. To KV it just looks like a pre-initialized, already running subscriber. It's completely agnostic to whether it's running, needs to be bounced, etc.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 57 at r20 (raw file):

It's got to do with the fact that when re-establishing the feed, we're doing so with higher and higher timestamps. I've changed how this automatic re-establishment of feeds works to leave it up to the caller to bounce the Subscriber as a whole. Added an assertion that checks that the initial timestamp we scan with is greater than the last recorded frontier timestamp. As for ensuring that frontier timestamps are monotonically increasing from the initial scan timestamp, there are assertions for it both at the rangefeed level and also at the rangefeed buffer level.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 86 at r20 (raw file):

Moved handlerInitialized out of the mutex, though still have handler in here after re-working things a bit. See above, curious to see how you feel about it now -- happy to get rid of the lazy binding if it's still messy. Or maybe even use another mutex (handlerMu).

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 171 at r20 (raw file):

See top-level comment, I think we can avoid all this complexity (that I originally introduced 😛).

Reviewed 3 of 3 files at r22, 4 of 5 files at r23, 1 of 1 files at r25, 23 of 25 files at r26, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 2015 at r20 (raw file):

Could you move the shouldQueue := up below this comment so that it's clear what it's referring to?

But also, should we be checking for each range whether we should queue? Something seems off about calling AdminN(1) and then queuing 10k ranges. That's especially true with applyAllFromSpanConfigStore.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):

The benefit is that we can avoid having to pass in the concrete type into pkg/kv/kvserver by having pkg/server.Start oversee the lifecycle of the Subscriber. To KV it just looks like a pre-initialized, already running subscriber. It's completely agnostic to whether it's running, needs to be bounced, etc.

Is this pre-initialization needed though? I guess I don't see why the Subscribe and OnSpanConfigUpdate methods both need to exist. I think this is what you mean by "fold a Start into the interface itself and disallow lazy bindings". Is the concern that you don't want KV managing the lifecycle of the subscriber?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 174 at r26 (raw file):

		log.Fatal(ctx, "currently subscribed: only allowed once at any point in time")
	}
	defer func() { atomic.CompareAndSwapInt32(&s.subscribed, 1, 0) }()

nit: isn't this just an atomic.StoreInt32(&s.subscribed, 0)?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 211 at r26 (raw file):

		})
		if err != nil {
			select {

Will these selects with a default clause allow the error to be silently ignored if the main goroutine is busy handling an event on one of the other channels?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 254 at r26 (raw file):

			mu.Unlock()

			select {

Same question here about the default case. I think this only works if the channel is buffered. But if you have multiple buffered channels then you can't make strong claims about which channel is read from first. Should this be reading from the ctx.Done() instead?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 275 at r26 (raw file):

	}
	defer rangefeed.Close()
	s.stopper.AddCloser(rangefeed)

Why do we need this? Aren't we already waiting on the stopper down below?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @andreimatei, @arulajmani, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 2015 at r20 (raw file):

Could you move the shouldQueue := up below this comment so that it's clear what it's referring to?

Done.

But also, should we be checking for each range whether we should queue? Something seems off about calling AdminN(1) and then queuing 10k ranges.

I was considering that, but this is actually exactly what we're doing with the system config span trigger above (since #53605) -- in that regard this is keeping the status quo. How would you feel about actually profiling once it's wired up before making any changes? It's hard to profile now given none of these tables are being written to, but we should be in much better shape next week after #71994 lands.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):

Yes, I didn't want KV managing the lifecycle of the subscriber; felt pkg/server feels a better home for it. The other thing is we can't import the kvsubscriber library in KV directly because of import cycles, so we'd have to pass in in through an interface. I felt it cleaner to not have aStart method on the interface where any mocked impl would have to block on some surrounding stopper.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 64 at r20 (raw file):

The number of restart attempts would be worth capturing in a metric, they're just logged prominently now. Is that what you were suggesting?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 211 at r26 (raw file):

Doh, good catch. I was originally trying to not have the handler thread block the underlying rangefeed, but that wasn't going to work here either. Changed the contract to have users ensure short-lived, non-blocking callbacks -- more than find for our purposes.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 254 at r26 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 275 at r26 (raw file):

Oops, don't need it. Removed.

This is getting close!

Reviewed 2 of 4 files at r27, 9 of 9 files at r28, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @andreimatei, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 2015 at r20 (raw file):

If this is the status quo then I'm ok with not changing behavior in this PR.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 64 at r20 (raw file):

Yes, the number of restarts would be a worthwhile metric. This can be a separate PR.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 192 at r28 (raw file):

	}

	return err // return the last error

Depending on the retryOpts, we don't always expect this to be reachable. Is that correct? Is so, consider alluding to that in a comment.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 257 at r28 (raw file):

		if err := buffer.Add(ctx, &bufferEvent{update, ev.Value.Timestamp}); err != nil {
			select {
			case <-ctx.Done():

I saw this pattern in one of Arul's PRs as well and found it very subtle that it doesn't deadlock. It relies on the contract written far away that a rangefeed callbacks' ctx is canceled when the rangefeed is closed.

Could you leave a comment near each of these case <-ctx.Done(): branches that mentions that we expect to hit this case after the handler goroutine stops listening to errCh and closes the rangeefed?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 310 at r28 (raw file):

	injectedErrCh := s.knobs.KVSubscriberErrorInjectionCh
	internal := spanconfigstore.New(s.fallback)
	var handlerInitialized bool

Could you confirm that we want to re-initialize the handler each time the rangefeed retries?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, and @nvanbenschoten)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 64 at r20 (raw file):

Added to #67679, hoping to burn through this list this month.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 257 at r28 (raw file):

Done. There's actually a stronger rangefeed API guarantee we're relying on. When invoking rangefeed.Close, we actually wait for the rangefeed library to ensure that all inflight, possibly blocking, handlers have finished running. See:

The stopped channel is closed before returning from rangefeed.run, the method actually invoking all the handlers one-by-one. Added this as a comment to rangefeed.Close.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 310 at r28 (raw file):

We do -- added a comment below.

This thing seems to be coming together! Mostly minor comments from me except for one. Can we reconsider the need to perform lazy binding here?

Reviewed 2 of 22 files at r13, 1 of 1 files at r14, 5 of 21 files at r15, 2 of 105 files at r17, 111 of 112 files at r18, 32 of 33 files at r21, 3 of 3 files at r22, 4 of 5 files at r23, 1 of 1 files at r24, 1 of 1 files at r25, 1 of 25 files at r26.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):
I have similar sentiments on this -- can we fold Subscribe and OnSpanConfigUpdate into a single method and put it on the interface instead? Something like:

type KVSubscriber interface {
   Subscribe(context.Context, retry.Options, func (update roachpb.Span))
}

This lets you forgo the complexity around if the handler has been lazily bound or not. I'm also not sure what lazily binding this handler gets us, considering we only ever allow a single handler.

I felt it cleaner to not have aStart method on the interface where any mocked impl would have to block on some surrounding stopper.

Now that you've pushed the retrying into the Subscribe call, would a mocked impl simply be able to return?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 186 at r27 (raw file):

	}{}

	defer func() {

This could use a comment

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 308 at r27 (raw file):

				}

				for _, ev := range events {

Do you only need this in the else case?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 317 at r27 (raw file):

			}
		case <-initialScanDoneCh:
			events := buffer.Flush(ctx, initialTS)

Might be worth calling out why we're using initialTS here as the timestamp to flush at, considering it's referencing behaviour in the rangefeed package.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 327 at r27 (raw file):

			s.mu.Unlock()

			if handler != nil {

Consider pulling this code out into a function so that you can call it both from here and above the very first time the handler is called.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 310 at r28 (raw file):

	injectedErrCh := s.knobs.KVSubscriberErrorInjectionCh
	internal := spanconfigstore.New(s.fallback)
	var handlerInitialized bool

This could use a comment

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 390 at r29 (raw file):

// OnSpanConfigUpdate installs a callback that's invoked with whatever span may
// have seen a config update.

nit: clarify that only one handler is ever allowed?

I'll (timidly) push back again against getting rid of lazy binding for the reasons outlined below, at best it's simplifying a nil check and a boolean and makes for what I think is a more awkward interface to mock out. The latter is something I'm trying to do in #71994, which is where the pushback stems from.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, and @nvanbenschoten)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):

Now that you've pushed the retrying into the Subscribe call, would a mocked impl simply be able to return?

No, it'll still have to a blocking call, right? Unless we want Subscribe to fire off an async task internally -- which is more surprising for the many reasons discussed elsewhere.

I'm also not sure what lazily binding this handler gets us, considering we only ever allow a single handler.

The lazy binding is not related to having a single handler, it's to do with not wanting to bunch together code dealing with the lifecycle the subscription and code that wants to learn about updates. The mock impl considerations are another aspect -- what does it mean to mock the interface that's providing you a retry option?

I still don't really see the complexity around lazy binding, everything else in this package was more intricate and that's where all my bugs were when writing tests. Really, it's one a nil check to see if a handler is available and a single bool.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 186 at r27 (raw file):

Added to lastFrontierTS.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 308 at r27 (raw file):

Yea, that works, done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 317 at r27 (raw file):

Renamed it to initialScanTS instead.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 327 at r27 (raw file):

Meh, it's two lines. I prefer the inline commentary.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 310 at r28 (raw file):

Added below + at the type level.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 390 at r29 (raw file):

Specified in the interface commentary.

Happy to chat about this offline, I may not be appreciating fully why we want to keep the lifecycle of subscription and learning about updates separate. I do think it buys us a bit more than simplifying a nil check though as it reduces the state this thing needs to track, which makes it easier to reason about.

The latter is something I'm trying to do in #71994, which is where the pushback stems from.

Are you referring to TestSpanConfigUpdateAppliedToReplica here or something else? If that's the one, then I think you could make it work even with the interface suggestion below.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):
I agree, there's not much code complexity around the lazy binding. To me the complexity stems from the fact that it isn't completely obvious to me why these distinct phases exist (as a first time reader of this code).

I could buy that you don't want to bunch together the code that deals with the lifecycle of the subscription and the code that wants to learns about updates, but then I'd expect us to be registering/de-registering multiple handlers while maintaining the rangefeed. We currently register only one handler and it is never deregistered. Do we see that changing in the future? If not, is there a benefit to having this separation?

No, it'll still have to a blocking call, right? Unless we want Subscribe to fire off an async task internally -- which is more surprising for the many reasons discussed elsewhere.

I wasn't suggesting that Subscribe fire off an async task internally. (I think) earlier you had retry code in the server package that retried this thing in a goroutine. Now that the retry code has been moved inside the kvsubscriber package, a mock implementation can simply no-op. I think I may have misunderstood what you meant above?

Yes, TestSpanConfigUpdateAppliedToReplica from that this branch is right. I'm still not following your suggestion, can you show what that test would look like if we did this differently?

I may not be appreciating fully why we want to keep the lifecycle of subscription and learning about updates separate

Because I feel they're separate things that benefit from the separation. I fear we're going in circles, I don't really have reasons other than ones we've already discussed below. Does this feel like a hard blocker? If not, I'll #meal-meetups expense you a beer for the rubber stamp.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, and @nvanbenschoten)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 246 at r16 (raw file):

We currently register only one handler and it is never deregistered. Do we see that changing in the future?

I don't see it changing in the future, no. If it does I'm happy to revisit.

a mock implementation can simply no-op. I think I may have misunderstood what you meant above?

But we want the handler to be called in the same goroutine as Subscribe, so it has to be blocking -- I'm not sure what "simply no-op" means. I'd welcome a code example of the test we're discussing above.

My comments are largely minor or superficial. This is looking good.

Minor/cleanup comments from me as well.

Reviewed 1 of 4 files at r27, 1 of 9 files at r28.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 37 at r34 (raw file):

// KVSubscriber is used to subscribe to global span configuration changes. It's
// a concrete implementation of the spanconfig.KVSubscriber interface.
//

It might be worth explicitly adding words around the lifecycle of this thing. Maybe a summary of what we discussed in yesterday's pod meeting to get here? Specifically, that we expect there to be a singleton KVSubscriber started per node that establishes and maintains the lifecycle of rangefeeds internally. It also allows all stores on that node to subscribe to to updates to particular spans by registering handlers.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 47 at r34 (raw file):

// updates and flushing them out en-masse in timestamp order when the rangefeed
// frontier is bumped[2]. If the buffer overflows (as dictated by the memory
// limit the KVSubscriber is instantiated with), the subscriber is wound down and

Should this say "subscriber retries" internally now?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 50 at r34 (raw file):

// an appropriate error is returned to the caller.
//
// When running into the errors above, it's safe for the caller to re-subscribe

This needs an update

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 67 at r34 (raw file):

//
// TODO(irfansharif): When swapping the old spanconfig.StoreReader for the new,
// instead of informing callers with an everything [min,max) span, we could diff

nit: s/callers/registered handlers/

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 75 at r34 (raw file):

//      getting split. When applying these updates, we need to make sure to
//      process the deletion event for S before processing S1...SN.
// [2]: In our example above deleting the config for S and adding configs for S1...Nwe

Nwe?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 79 at r34 (raw file):

//      lest we expose the intermediate state where the config for S was deleted
//      but the configs for S1...SN were not yet applied.
// [3]: TODO(irfansharif): When tearing down the subscriber due to underlying errors,

Now that we're doing this internally, should we address this TODO? Or, if not, move it inline?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 173 at r34 (raw file):

//      the exported StoreReader will be up-to-date and continue to be
//      incrementally maintained.
func (s *KVSubscriber) Subscribe(ctx context.Context) error {

In line with Andrew's comment about renaming OnSpanConfigUpdate to Subscribe, let's rename this thing to Start?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 199 at r34 (raw file):

• 1 on Start / run pattern we have in other places as well.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 217 at r34 (raw file):

	defer func() {
		mu.Lock()
		s.lastFrontierTS = mu.frontierTS

Should this be a s.lastFrontierTS.Forwad(mu.frontierTS)instead?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 375 at r34 (raw file):

• 1

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber_test.go, line 17 at r34 (raw file):

// TestingSubscribeInner exports the inner subscription route for testing
// purposes.
func (s *KVSubscriber) TestingSubscribeInner(ctx context.Context) error {

Move this method to a testutils file instead.

What in this change obviates the need to do this?

This was a stand alone commit unrelated to anything else in the PR. It's actually just dead code like mentioned in the commit message. Was discussed in this thread: https://reviewable.io/reviews/cockroachdb/cockroach/69614#-MlLiZryEFeTuS3XXoq2.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, and @nvanbenschoten)

pkg/kv/kvserver/store.go, line 2010 at r36 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 37 at r34 (raw file):

I feel like how this thing is expecting to get used is not commentary well suited here. A lot of the text below talks about the lifecycle, and looking the uses it's (I hope) clear how it's being used -- "started at the node level once" and "subscribed to by each store".

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 47 at r34 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 50 at r34 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 75 at r34 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 79 at r34 (raw file):

I'm fine with just leaving the TODO here.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 173 at r34 (raw file):

Already done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 178 at r34 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 199 at r34 (raw file):

Already done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 217 at r34 (raw file):

Sure, done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 375 at r34 (raw file):

Already done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 188 at r36 (raw file):

Done (was already mentioned in run).

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber_test.go, line 17 at r34 (raw file):

Does this apply? I'm exporting a package internal method.

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @ajwerner, @arulajmani, @irfansharif, and @nvanbenschoten)

pkg/spanconfig/spanconfig.go, line 51 at r37 (raw file):

// callers can subscribe to learn about what key spans may have seen a
// configuration change. After learning about a span update through a callback
// invocation, callers can consult the embedded StoreReader to retrieve an

s/callers/subscribers/ (in a couple of places below as well).

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 37 at r34 (raw file):

IMO the details about the lifecycle are getting lost in some of the details below. I'd strongly encourage you to write about the two phases (start and subscribe) and why they exist, even if you do it in general terms rather than talking about specific callers :)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 79 at r34 (raw file):

Okay. It's still worth updating the TODO to talk about internally retrying from the last saved checkpoint as opposed to surfacing it to the callers of this function.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 97 at r37 (raw file):

	knobs               *spanconfig.TestingKnobs

	subscribed int32    // accessed atomically

s/subscribed/started

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber_test.go, line 17 at r34 (raw file):

I would've expected kvsubscriber_test.go to have unit tests and testutils.go to have wrappers around internal methods we want to export for testing purposes.

Thanks y'all! I'm sure this felt like a doozy.

bors r+

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @ajwerner, @arulajmani, and @nvanbenschoten)

pkg/spanconfig/spanconfig.go, line 51 at r37 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 37 at r34 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 79 at r34 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 97 at r37 (raw file):

Done.

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber_test.go, line 17 at r34 (raw file):

It's not atypical to export test-only symbols in a _test package, which is why I did it this way (testutils.go would surface this symbol in the godoc and your editor unless it's testutils_test.go 🤷‍♀️).

Reviewed 1 of 2 files at r29, 6 of 10 files at r35, 3 of 4 files at r37, 2 of 2 files at r38, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (and 1 stale)

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 336 at r38 (raw file):

			events := buffer.Flush(ctx, frontierTS)
			s.mu.Lock()

It would have been nice to have added some commentary about this locking. It's pretty subtle, but as we've found, also very important. Maybe we can add a comment next time we're working around here?

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 418 at r38 (raw file):

}

type handler struct {

Much nicer!

pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go, line 428 at r38 (raw file):

		h.initialized = true

		if update.Equal(keys.EverythingSpan) {

Out of curiosity, why can't we always return on this path? Do we ever need the second call to fn if we just called it with keys.EverythingSpan?