jschoubben
09aa063a13
mesh/merge-gate out of date: judged against trunk 5be1cfaed19b, the trunk is now f10fbda1f619; judged again on a recheck or when it is merged
mesh/repo-check out of date: judged against trunk 5be1cfaed19b, the trunk is now f10fbda1f619; judged again on a recheck or when it is merged
mesh/delivery delivered
A declaration is one bus message and the control node's passed the limit by 2 KB. The size is checked at every send against the connection's own limit, with a condition from 90% and a message naming size and limit when over; the export, 138 KB compact of 165 KB indented, is sent compact.
87 lines
3.2 KiB
Go
87 lines
3.2 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
)
|
|
|
|
// Signer is whatever holds the control plane's signing key.
|
|
type Signer interface {
|
|
Sign(ctx context.Context, message []byte) ([]byte, error)
|
|
}
|
|
|
|
// ActingGate is what every send passes before it is made (novox/hq to-be 45 §6): whether this process
|
|
// may act under the controller's lease. Set by the controller; nil passes every send — a test, a tool.
|
|
var ActingGate func(ctx context.Context) error
|
|
|
|
// Declare sends a node what it should be, signed.
|
|
//
|
|
// The signature is over the declaration exactly as it is published — the same bytes the node
|
|
// verifies. Anything that re-encoded between here and there would produce a signature over
|
|
// something else, and the node would refuse a declaration that was genuinely the mesh's.
|
|
//
|
|
// Published to the node's own queue, which its account alone may read.
|
|
func Declare(ctx context.Context, bus Bus, signer Signer, node string,
|
|
declaration []byte, timeout time.Duration) error {
|
|
|
|
if !json.Valid(declaration) {
|
|
return fmt.Errorf("refusing to send %s something that is not a declaration", node)
|
|
}
|
|
// **At the send, not only where it was composed**: a lease lost between the two stops this one.
|
|
if ActingGate != nil {
|
|
if err := ActingGate(ctx); err != nil {
|
|
return fmt.Errorf("%s was not sent its declaration: %w", node, err)
|
|
}
|
|
}
|
|
|
|
signature, err := signer.Sign(ctx, declaration)
|
|
if err != nil {
|
|
return fmt.Errorf("cannot sign a declaration for %s: %w", node, err)
|
|
}
|
|
|
|
body, err := json.Marshal(Signed{Declaration: declaration, Signature: signature})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// **Measured before it is sent** (novox/hq issue 1689): the bare bus error said only "maximum payload
|
|
// exceeded", on 2026-10-21 for the control node's declaration, 2 KB over. The limit is the connection's own.
|
|
if limit := payloadLimitOf(bus); limit > 0 {
|
|
size := int64(len(body))
|
|
if DeclarationMeasured != nil {
|
|
DeclarationMeasured(ctx, node, size, limit)
|
|
}
|
|
if size > limit {
|
|
return fmt.Errorf("%s's declaration is %d bytes on the wire (%d bytes of declaration, signed and encoded), "+
|
|
"over the bus's largest message of %d bytes by %d: it cannot be sent (novox/hq issue 1689)",
|
|
node, size, len(declaration), limit, size-limit)
|
|
}
|
|
}
|
|
|
|
publish, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
return bus.PublishDeclaration(publish, node, body)
|
|
}
|
|
|
|
// NearTheLimit is the share of the bus's largest message at which a declaration is said to be near it.
|
|
const NearTheLimit = 0.9
|
|
|
|
// DeclarationMeasured is told, at every send of a declaration, its size on the wire and the bus's largest message,
|
|
// so the controller raises a condition when it nears the limit and clears it when it is back under (issue 1689).
|
|
// Set by the controller; nil in a test or a tool.
|
|
var DeclarationMeasured func(ctx context.Context, node string, size, limit int64)
|
|
|
|
// payloadLimiter is a bus that knows its largest message.
|
|
type payloadLimiter interface{ MaxPayload() int64 }
|
|
|
|
// payloadLimitOf is the bus's largest message in bytes, or 0 when the bus does not say.
|
|
func payloadLimitOf(bus Bus) int64 {
|
|
if l, ok := bus.(payloadLimiter); ok {
|
|
return l.MaxPayload()
|
|
}
|
|
return 0
|
|
}
|