Let two controllers overlap safely while one hands over to the other (hq issue 213)
The controller's machine moves it from the container to a process by starting the process first and removing the container once the process is up (mesh-host's `replaces`). For that moment two controllers share the store and the bus. Checked what each does: - the seat's verbs: a queue group per seat, each call answered once. Safe. - the controller's consumers on CONTROL and EVENTS: push consumers with no delivery group, so the second bind is refused with "consumer is already bound" and serve exited. The process would restart for ever, the host would never see it up, and the container would never go. The second controller now stands by and binds when the first lets go (tested on a real bus; fails without the change). - plans: read, changed and saved whole by the 30s timer, by build outcomes, by a merge and by `plans stop`. Two timers would each ask a tier the other had just asked. Working the plans now takes a session-level advisory lock on the inventory: the timer skips while another holds it, the other paths wait for it. Build asks happen only inside plan work and are covered by the same lock.
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -78,10 +79,15 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
|
||||
// than creating one here: the consumer is an object with a configuration — ack policy, ack
|
||||
// wait, redelivery — and a client that creates its own would be a second opinion about it.
|
||||
control := make(chan *nats.Msg, Prefetch)
|
||||
said, err := js.ChanSubscribe("", control, nats.Bind("CONTROL", broker.ControllerName))
|
||||
said, err := standingBy(ctx, log.Default(), "CONTROL", func() (*nats.Subscription, error) {
|
||||
return js.ChanSubscribe("", control, nats.Bind("CONTROL", broker.ControllerName))
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("subscribing to what nodes say: %w", err)
|
||||
}
|
||||
if said == nil {
|
||||
return nil // stopped while standing by
|
||||
}
|
||||
defer func() { _ = said.Unsubscribe() }()
|
||||
|
||||
// Heartbeats, on core NATS and off any stream (design 25 §3). Their own subscription because
|
||||
@@ -97,10 +103,15 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
|
||||
var events chan *nats.Msg
|
||||
if len(n.follows) > 0 {
|
||||
events = make(chan *nats.Msg, Prefetch)
|
||||
followed, err := js.ChanSubscribe("", events, nats.Bind("EVENTS", broker.ControllerName))
|
||||
followed, err := standingBy(ctx, log.Default(), "EVENTS", func() (*nats.Subscription, error) {
|
||||
return js.ChanSubscribe("", events, nats.Bind("EVENTS", broker.ControllerName))
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("subscribing to what the catalogue says: %w", err)
|
||||
}
|
||||
if followed == nil {
|
||||
return nil
|
||||
}
|
||||
defer func() { _ = followed.Unsubscribe() }()
|
||||
}
|
||||
|
||||
@@ -324,3 +335,44 @@ func (m *natsControl) forget() {
|
||||
type replyAddressed struct {
|
||||
ReplyTo string `json:"reply_to,omitempty"`
|
||||
}
|
||||
|
||||
// StandbyPoll is how often a controller standing by looks again for its consumers. A variable so a
|
||||
// test need not wait.
|
||||
var StandbyPoll = 2 * time.Second
|
||||
|
||||
// standingBy binds one of the controller's consumers, waiting while another controller holds it.
|
||||
//
|
||||
// **Two controllers, one consumer** (novox/hq issue 213). The controller's consumers are push
|
||||
// consumers with no delivery group, so the server lets one subscription bind each — on purpose:
|
||||
// two would each act on every message (issue 146). When a machine hands its controller over from
|
||||
// the container to the process, the host starts the process first and removes the container only
|
||||
// once the process is up; the process then finds the consumers bound. Exiting on that would never
|
||||
// be up, so the container would never go. It stands by instead — the seat's verbs are already
|
||||
// served from a queue group, and the plans wait on their lock — and binds as soon as the other lets
|
||||
// go. Nil and no error is ctx ending while it waited.
|
||||
func standingBy(ctx context.Context, logger interface{ Printf(string, ...any) }, stream string,
|
||||
bind func() (*nats.Subscription, error)) (*nats.Subscription, error) {
|
||||
said := false
|
||||
for {
|
||||
sub, err := bind()
|
||||
if err == nil {
|
||||
if said {
|
||||
logger.Printf("took the controller's consumer on %s: the controller that held it let go", stream)
|
||||
}
|
||||
return sub, nil
|
||||
}
|
||||
if !strings.Contains(err.Error(), "already bound") {
|
||||
return nil, err
|
||||
}
|
||||
if !said {
|
||||
logger.Printf("another controller holds the controller's consumer on %s; standing by "+
|
||||
"until it lets go", stream)
|
||||
said = true
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, nil
|
||||
case <-time.After(StandbyPoll):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user