package link import ( "context" "crypto/ed25519" "encoding/json" "errors" "fmt" "time" ) // Membership is what a node needs to reach its mesh again, held by the caller. type Membership struct { Node string Broker string Fingerprint string Password string Signer ed25519.PublicKey // Transport is which bus this membership was minted for (hearing.go): the mesh's own, and a // membership that names another is one this host cannot dial with. Transport string } // Announce is how the link says what is happening, so a node running unattended leaves an // account of it. Nil is allowed and means say nothing. type Announce func(string) // Hold keeps this node in the mesh, reconnecting for as long as it is asked to. // // Disconnection is an ordinary situation and not a failure (novox/hq ADR 0004), so this does not // give up. A laptop shut for a week comes back and reconnects; it does not come back needing // somebody to start it again. // // The backoff exists because the two common reasons differ in how long they last: a broker // restarting is back in seconds, and a machine that has moved to a network with no route may be // hours. Retrying every second for hours is a node shouting into nothing; waiting a minute after // a broker blip is a node that is needlessly late. So it starts fast and slows down, and resets // once a connection has actually held. // Roused is a channel that says the machine has reason to believe its link is stale — it woke // from suspend, or its network changed. // // **The machine knows before any timeout does.** A suspended laptop's connection is dead the // moment it wakes, and heartbeats find that out in twenty or thirty seconds; for that time the // node believes it is in the mesh and is not, which is the one state this design says must never // be indistinguishable from being connected. Nothing new listens on the node to arrange it — the // signal a service manager already sends is enough (novox/hq ADR 0004). // // Nil is allowed and means nothing ever rouses it, which is every machine that does not suspend. type Roused <-chan struct{} func Hold(ctx context.Context, m Membership, queue *Session, say Announce, timeout time.Duration) error { return HoldRoused(ctx, m, queue, say, timeout, nil) } // HoldRoused is Hold, told when the machine has reason to think its link is stale. // // The session outlives every link and says the node's facts on whichever is open (novox/hq ADR 0539 §9). func HoldRoused(ctx context.Context, m Membership, queue *Session, say Announce, timeout time.Duration, roused Roused) error { return holdWith(ctx, func(ctx context.Context) error { return Run(ctx, m, queue, say, timeout) }, say, roused) } // attempt is one try at holding the link open, returning when it ends for any reason. // // Named so the loop below can be driven without a broker. What the loop decides — when to wait, // how long, what being roused does — is the part with the reasoning in it, and it was reachable // only through a real connection before. type attempt func(context.Context) error func holdWith(ctx context.Context, run attempt, say Announce, roused Roused) error { const ( first = 2 * time.Second most = 2 * time.Minute // A connection that lasted this long counts as having worked, so the next failure starts // from the bottom again. Without it a node that reconnects and immediately drops climbs // to the maximum and stays there, long after whatever caused it went away. settled = 30 * time.Second ) wait := first for { began := time.Now() // The link runs under a context this loop can cancel, so being roused ends the current // attempt rather than only shortening the wait after it. // // **That is the whole of it.** After a resume the socket looks perfectly healthy from // inside this process — there is no error and no close, because nothing has tried to // send anything. It is heartbeats that eventually discover it, twenty or thirty seconds // later. A machine that knows it just woke does not have to wait to be told. // // A rouse that turns out to be spurious costs one reconnect, which is cheap and // idempotent: the node redeclares its queue and anything unacknowledged is redelivered. // The alternative costs half a minute of believing it is in a mesh it has left. trying, done := context.WithCancel(ctx) if roused != nil { go func() { select { case <-trying.Done(): case <-roused: say("woken or moved — dropping the link and opening it again") done() } }() } err := run(trying) done() if ctx.Err() != nil { return nil } if time.Since(began) > settled { wait = first } switch { case errors.Is(err, ErrWrongCertificate): // Said in full every time rather than folded into a retry count. This does not mean // the network is down; it means what answered is not the mesh this node joined, and // no amount of waiting fixes it. The node keeps running what it was last told, which // is the right thing to do while somebody works out what happened. say("the broker is not the one this node joined: " + err.Error()) say("this will not fix itself. This node keeps running what it was last told.") case err != nil: say(fmt.Sprintf("disconnected: %v — trying again in %s", err, wait)) default: say(fmt.Sprintf("the link closed — trying again in %s", wait)) } select { case <-ctx.Done(): return nil case <-roused: // And it does not serve out a wait computed for a broker that was restarting, either. // // The backoff is not *reset* by this. Being roused says the machine changed, not that // whatever was refusing the connection has stopped — a laptop woken repeatedly on a // network with no route would otherwise retry at full speed for as long as somebody // keeps opening the lid. say("woken or moved — trying again now") case <-time.After(wait): } if wait *= 2; wait > most { wait = most } } } // Run holds the link open once, hearing what the controller asks and saying the node's facts. // // Outbound only, and nothing listens on this machine. Returns when the link ends, for any reason; // Hold is what decides whether to open it again. func Run(ctx context.Context, m Membership, queue *Session, say Announce, timeout time.Duration) error { link, err := Open(ctx, m, timeout) if err != nil { return err } defer link.Close() return serve(ctx, link, m, queue, say, timeout) } // serve is Run on a link already open: separated so what the node says, and when, can be tested // without a broker. func serve(ctx context.Context, link Link, m Membership, queue *Session, say Announce, timeout time.Duration) error { if say == nil { say = func(string) {} } // Said, because it is the event anybody watching actually wants. Without it a node logs every // failure and nothing on success, so a log full of "trying again" and then silence reads as // still broken when it means the opposite. say("in the mesh, hearing what the controller asks of this node") // The session says this node's facts as the link is handed to it, and is not let go while it is mid-act. queue.Attach(ctx, link) defer queue.Detach(link) // And the mesh asking this node to say its facts again, on a link that hears the question. var asks <-chan struct{} if a, ok := link.(Asked); ok { asks = a.AskedToReport() } // And the controller asking this node to hand a directory used as found to the mesh (novox/hq issue // 356), answered on the link the question came over. if h, ok := link.(HandOvers); ok { h.HearHandOvers(queue.HandOver) } // And asking for a fresh search for setuid programs (novox/hq issue 361). if s, ok := link.(SetuidSearches); ok { s.HearSetuidSearches(queue.SetuidSearch) } // And asking for a read-only judgement of its directories (novox/hq issue 358). if j, ok := link.(DirectoryJudgements); ok { j.HearDirectoryJudgements(queue.JudgeDirectories) } // And asking for the take act on a library, on a warrant (novox/hq ADR 0328 §7). if s, ok := link.(AccessActs); ok { s.HearAccessActs(queue.AccessAct) } // And asking to move a value sealed to this machine into the vault (design 57's V3, issue 676). if s, ok := link.(MoveSecrets); ok { s.HearMoveSecrets(queue.MoveSecret) } for { if ctx.Err() != nil { return nil } select { case <-ctx.Done(): return nil case reason := <-link.Lost(): return reason case <-asks: queue.FactsAsked(ctx) } } } // publishReport tells the mesh what this node did, and says whether the broker took it. // Publish sends one report on this node's own connection and returns: the one-shot path for a // report a command makes rather than the running host — a rekey (novox/hq ADR 0105). The same // account, the same pinned certificate and the same exchange as the running host's reports. func Publish(ctx context.Context, m Membership, report Report, timeout time.Duration) error { link, err := Open(ctx, m, timeout) if err != nil { return err } defer link.Close() var said string if !publishReport(ctx, link, m, report, func(s string) { said = s }, timeout) { return errors.New(said) } return nil } func publishReport(ctx context.Context, bus Bus, m Membership, report Report, say Announce, timeout time.Duration) bool { report.Node = m.Node body, err := json.Marshal(report) if err != nil { say("cannot encode this node's own report: " + err.Error()) return false } publish, cancel := context.WithTimeout(ctx, timeout) defer cancel() // Said rather than swallowed. A report that fails to publish leaves the mesh believing this // node never answered, while the node believes it did — and the two would go on disagreeing // with nothing anywhere saying so. That shape of fault is the one this project keeps finding. if err := bus.Report(publish, m.Node, body); err != nil { say(fmt.Sprintf("applied, and could not tell the mesh: %v", err)) return false } return true }