I wonder if it makes sense for the ingestion processor to be aware of the concept of generations at all. Would it be simpler if we just swallowed ErrBadConn in the driver in the client which would hang the processor until its context was cancelled.

Also does errors.Is(err, driver.ErrBadConn) work here? I think that's the more standard way of checking for errors (see https://github.com/cockroachdb/cockroach/blob/master/docs/RFCS/20190318_error_handling.md for more info).

The processor would go to a different state If we swallow ErrBadConn on the client side, since there are no more values to be read from eventCh, consequently we'd stop reading from cutoverCh. I'm not sure if there's a better way of differentiating the error returned by the client.

@pbardea If I'm understanding correctly, the issue with swallowing the error is that we return from the sinkless client. The return then triggers the eventCh for the partition to be closed, which in turn causes the merged channel in the processor to be closed. Once the merged channel is closed, the proc drains and we're in a bit of a soup.

Thinking ahead to multi-partition where we probably want to do more than just wait for a cutover (ingest up until some time), a generation event would give us a place to trigger this custom logic? There might be a better way though

Gotcha, my original thinking was that we wouldn't close the eventCh and the processors would be unaware that the client had disconnected, it would just hang waiting for its context to be canceled. (When the client sees the error it just reads from <-ctx.Done().)

The tl;dr of below is that given that the coordinator doesn't yet have a way to send processors messages, this approach should be okay for now, but I think it's worth keeping an eye on potentially moving this to the coordinator down the line.

A bit more consideration to why it may be good to move to a scheme where the coordinator is the only one aware of "cutovers times" and "generation events":

Such a scheme would allow each ingestion processor to not need to worry about global state like generations and cutover time, but only to ingest data from the stream, and to watch for a "ingest until at least " signal from the coordinator. (We don't have a way for the coordinator to provide a signal like that today).

The underlying concern is that on large clusters, we'll develop these many to 1 relationships that could put unnecessary pressure on some nodes. (We have this today with cutover time polling since all 100 nodes would poll the job record on avg 3 times a second and that would only increase as the polling interval is reduced).

100% agree, filed #68475.

I'd prefer if instead of this bool we just:

<-sip.cutoverCh
sip.internalDrained = true
return nil, nil

from below. This is also most likely going to change in the future when in multi-partition streaming we want to continue reading events up until some time in the future (as indicated by the GenerationEvent)?

We'd probably also want to wait on ctx.Done() in addition to cutover channel here.

I updated the logic here, hope it makes sense now. When a generation event is received, we wait for sip.cutoverCh and sip.Ctx.Done(), whichever comes first. If we received value from sip.cutoverCh, the function returns. If the context gets cancelled before a cutover signal is sent, the function returns as well.

I like the idea of logging a generation event. This will be more useful when GE actually contains a generation time or something for multi parititon. For the time being lets change the verbiage to "GenerationEvent received" since client disconnected is specific to the sinkless client.

I don't think we need any of this once we address the comment below?

mmm this seems a little wrong, it should probably be:

if err := rows.Err(); err != nil {
  // Maybe add a comment about why we are special casing a client disconnect.
  if errors.Is(err, driver.ErrBadConn) {
       select {
          case eventCh <- streamingccl.MakeGenerationEvent():
          case <-ctx.Done():
				       errCh <- ctx.Err()
			 }
  } else {
      errCh <- err
  }
}

Let's chat offline about how select blocks and channels work, just to make sure we're on the same page 🙂

I haven't reviewed the test yet, and I'm going to think if there is any other way we can structure this. I don't particularly like sleep in tests 😋 time-dependent tests often flake and get skipped

Perhaps instead of testing if the client hangs, we could test if the processor returns an error or not when the client disconnects?

We need to move the wg.Wait() above this so we are guaranteed that sip.Run()` has returned.

can we then use out returned by getStreamIngestionProcessor to check:

if !out.ProducerClosed() {
		t.Fatalf("output RowReceiver not closed")
}

for {
			row := out.NextNoMeta(t)
			if row == nil {
				break
			}
      // We don't expect any rows so...
      t.Fatal(...)
}

Update comment to:
// The sip processor has received a gen event and is thus waiting for a cutover signal, so let's send one!

can we pass a func() as a parameter here instead of a channel? This will give us flexibility in the future to add stuff to that func.

Maybe also rename the function to makeGenerationEventReceived

I don't think we need this any longer. out is a row buffer that is filled by sip.Next(). The out.NextNoMeta checks that none of the output was a meta.

we need to sip.mu.Lock() and sip.mu.Unlock() the mutex around this.

ditto comment as above

we usually use syncutil.Mutex

@annezhu98 one nit, though you can pick it up in your next PR to prevent another push since this is only testing code. We should close(errCh) too.