Pin mesh-sdk at the group's head acbf80a in every module the group touches, so the provisioner carries what a consumer contributes and route-proxy keeps its routes; count a removal confirmed only when the bus says the stream or consumer is not found (issue 774)
mesh/delivery merged: source.main moved to novox/mesh-catalog@bcbff9de365c
mesh/delivery-group group feat/072-moves merged: every member is merged
mesh/check failed: shanks: the check of novox/mesh-controller#487 failed: go mod download in modules/route-proxy: go: git.novox.be/novox/mesh-controll…
mesh/delivery merged: source.main moved to novox/mesh-catalog@bcbff9de365c
mesh/delivery-group group feat/072-moves merged: every member is merged
mesh/check failed: shanks: the check of novox/mesh-controller#487 failed: go mod download in modules/route-proxy: go: git.novox.be/novox/mesh-controll…
This commit is contained in:
@@ -121,9 +121,13 @@ func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) (
|
||||
sctx, cancel = step(ctx)
|
||||
_, err = js.Consumer(sctx, stream, consumer)
|
||||
cancel()
|
||||
// Only the bus saying it is not there is a removal; a read that failed otherwise confirms nothing.
|
||||
if err == nil {
|
||||
return nil, fmt.Errorf("consumer %s of %s is still there after the removal was accepted", consumer, stream)
|
||||
}
|
||||
if !errors.Is(err, jetstream.ErrConsumerNotFound) && !errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return nil, unanswered(err, fmt.Sprintf("the removal of consumer %s of %s was accepted but could not be confirmed", consumer, stream))
|
||||
}
|
||||
out := ConsumerRemoved{Stream: stream, Consumer: consumer, Why: w, Was: was, Removed: true, At: time.Now().UTC()}
|
||||
if err := said(consumerRemoved, out); err != nil {
|
||||
return nil, fmt.Errorf("consumer %s of %s WAS removed (%s) but the audit event could not be said: %w", consumer, stream, w, err)
|
||||
@@ -295,9 +299,13 @@ func RemoveStream(ctx context.Context, bus Bus, said Said, a map[string]any) (an
|
||||
sctx, cancel = step(ctx)
|
||||
_, err = js.Stream(sctx, stream)
|
||||
cancel()
|
||||
// Only the bus saying it is not there is a removal; a read that failed otherwise confirms nothing.
|
||||
if err == nil {
|
||||
return nil, fmt.Errorf("stream %s is still there after the removal was accepted", stream)
|
||||
}
|
||||
if !errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return nil, unanswered(err, fmt.Sprintf("the removal of stream %s was accepted but could not be confirmed", stream))
|
||||
}
|
||||
out.Removed = true
|
||||
out.Said = fmt.Sprintf("%s removed: %d messages (%d bytes) and no consumer", stream, out.Messages, out.Bytes)
|
||||
if err := said(streamRemoved, out); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user