queue: when should a Processor's context be cancelled?

[Zaba505]

Note

This isn't a proposal. It's a place to go back and forth on how a Processor's context should end, before anything is proposed. It came up in Proposal: Kafka processor walkthrough, whose consumer now retries through a retry topic.

Today

A Processor's context, for Kafka and RabbitMQ alike:

• keeps Serve's values but not its cancellation. When shutdown starts, the record in flight goes on, so a Processor that honours its context doesn't fail it on every SIGTERM, which under AtLeastOnce would make every rollout exit 1 (Kafka, RabbitMQ);
• is cancelled when the grace period runs out. The App then exits 1 with a *humus.GracePeriodError, and the record isn't committed or acked, so it's processed again (Kafka, RabbitMQ);
• isn't cancelled by a Kafka revoke. The revoke waits for the record in flight, and a member whose revoke outlasts RebalanceTimeout, 60s by default, is fenced out of the group (Revocation).

A Process call ends one of two ways: nil commits or acks the record, and an error under AtLeastOnce stops the Service, with exit 1. There is no way to hand a record back, neither done nor failed.

HTTP and gRPC handlers work the same way: a request's context doesn't derive from Serve's, and ends when the grace period does (HTTP, gRPC).

Two tests in queue/kafka show it: the context is still live after SIGTERM, and it is cancelled exactly when the grace period runs out.

Where it bites

#950's processor sends a record that fails for a reason that isn't its own to orders.retry, with its attempt count in a header. One Process handles both topics. It waits before a retried record, for a time derived from the record's attempt count, as long as that is:

func (p *projector) Process(ctx context.Context, m kafka.Message) error {
	select {
	case <-time.After(time.Until(m.Timestamp.Add(p.backoff(attempt(m))))):
	case <-ctx.Done():
		return ctx.Err()
	}
	return p.project(ctx, m)
}

In a healthy system that's fine. It bites when the App shuts down, or its partitions move, while a downstream service is out and a record is waiting:

• a SIGTERM waits out the whole grace period, and the rollout ends in exit 1, though nothing is lost;
• a revoke waits up to RebalanceTimeout, and then the member is fenced out;
• the Processor can't tell that either has started, and if it could, it has no way to give the record back that isn't a failure.

The same holds for any Processor blocked on a downstream that is out, such as one retrying with backoff inside Process.

To go back and forth on

• Should a Processor learn that shutdown has started? By its context being cancelled, or by a signal it can choose to watch, so that a record almost done still finishes?
• Should it learn of a revoke the same way?
• Should a record be able to be handed back, neither committed nor failed? Kafka's next owner would process it again from the last commit. RabbitMQ would requeue it, as it requeues the deliveries a Service received but didn't process.
• When records were cut short but none was lost, is exit 1 the right outcome?
• Should queue Processors keep matching HTTP and gRPC handlers here, or differ, since a record can be given back and a request can't?