package inventory import ( "context" "errors" "fmt" "sort" "time" "github.com/jackc/pgx/v5" "github.com/novox/mesh-controller/internal/chain" ) // The links of the chain (novox/hq ADR 0408 §1, design 47 acceptance 10, issue 1095): a send that puts a build on a machine // writes, in the assignment of the module there, the build, what it published, its commit and the rollout that sent it, and // since when; the rollout lists the assignments it changed with the build before and after (the controller's send, in // release_plan.go). One read of an assignment then names the commit, the build and the rollout that put it there, and one read // of the rollout lists its assignments. type rolloutKey struct{} // RolloutModule is what a rollout sends of one module: the commit and the build of it. type RolloutModule struct{ Commit, Build string } type rolloutOf struct { id string modules map[string]RolloutModule } // WithRollout is a context for a send a rollout makes: what RecordSent writes in the assignments of the modules it carries // names this rollout and the build it sent. A module the send carries beside these, that another walk is moving, is named // by the build the mesh holds of its commit and by no rollout. func WithRollout(ctx context.Context, id string, modules map[string]RolloutModule) context.Context { return context.WithValue(ctx, rolloutKey{}, rolloutOf{id: id, modules: modules}) } // AssignmentRuns is what each module assigned to a machine runs there, as its assignment names it: the build and the // commit. Empty when the assignments are not on the bus, and for a module not sent yet. func (i *Inventory) AssignmentRuns(ctx context.Context, node string) (map[string]chain.Ref, error) { c, err := i.theChain(ctx) if err != nil { return nil, err } all, err := c.Assignments.All(ctx) if err != nil { return nil, err } out := map[string]chain.Ref{} for _, k := range all { a := k.Value if a.Node == node && a.Stands() && (a.Build != "" || a.Commit != "") { out[a.Module] = chain.Ref{Build: a.Build, Commit: a.Commit} } } return out, nil } // AssignmentsOn are the assignments of a machine as the chain names them, by module: the standing ones, with their links. func (i *Inventory) AssignmentsOn(ctx context.Context, node string) ([]chain.Assignment, error) { c, err := i.theChain(ctx) if err != nil { return nil, err } all, err := c.Assignments.All(ctx) if err != nil { return nil, err } var out []chain.Assignment for _, k := range all { if a := k.Value; a.Node == node && !a.Assigned.IsZero() && a.Stands() { out = append(out, a) } } sort.Slice(out, func(a, b int) bool { return out[a].Module < out[b].Module }) return out, nil } // recordRuns writes what a send put on a machine into the assignments, once the send is recorded, and it is how the mesh // knows what a machine was sent (SentBuilds). **A send that names its builds** writes, for each module it carried a build of // — by the commit — the build, what it published, the commit, the rollout and since when, on the module's key: filled where // the key names nothing, replaced when the commit moved, never blanked by a push by hand; a module nothing assigns, carried // because another requires it, gets a standing key marked pulled-in; and every other key of the machine loses what it named, // for the machine is not sent it any more. **A send that names none** (a declaration by hand) makes what the machine runs // unknown: every key of it says so (sent-unknown) and names no commit. **An error here fails the record of the send**, for // what a machine was sent is read from these keys and from nowhere else. func (i *Inventory) recordRuns(ctx context.Context, nodeID string, builds map[string]string, at time.Time) error { c, err := i.theChain(ctx) if err != nil { return fmt.Errorf("what the machine runs cannot be written: %w", err) } var node string if err := i.store.Pool().QueryRow(ctx, `select name from node where id = $1`, nodeID).Scan(&node); errors.Is(err, pgx.ErrNoRows) { return nil } else if err != nil { return err } rollout, _ := ctx.Value(rolloutKey{}).(rolloutOf) all, err := c.Assignments.All(ctx) if err != nil { return fmt.Errorf("the assignments of %s cannot be read: %w", node, err) } var failed []error for _, k := range sortedValues(all) { a := k.Value if a.Node != node || !a.Stands() { continue } if builds[a.Module] != "" { continue // carried, and its commit is written below } if err := i.forgetRun(ctx, c, k, builds == nil, at); err != nil { failed = append(failed, fmt.Errorf("%s on %s: %w", a.Module, node, err)) } } modules := make([]string, 0, len(builds)) for m := range builds { modules = append(modules, m) } sort.Strings(modules) for _, m := range modules { if builds[m] == "" { continue } if err := i.recordRun(ctx, c, node, m, builds[m], rollout, at); err != nil { failed = append(failed, fmt.Errorf("%s on %s: %w", m, node, err)) } } return errors.Join(failed...) } func sortedValues(all map[string]chain.Kept[chain.Assignment]) []chain.Kept[chain.Assignment] { out := make([]chain.Kept[chain.Assignment], 0, len(all)) for key, k := range all { k.Key = key out = append(out, k) } sort.Slice(out, func(a, b int) bool { return out[a].Key < out[b].Key }) return out } // forgetRun writes that a machine is no longer sent what a key named: its links are cleared, and for a send by hand the key // says what the machine runs is unknown. func (i *Inventory) forgetRun(ctx context.Context, c *chain.Chain, k chain.Kept[chain.Assignment], hand bool, at time.Time) error { for try := 0; try < casTries; try++ { next := k.Value next.Build, next.Artifact, next.Package, next.Commit, next.Rollout, next.Since = "", "", nil, "", "", time.Time{} next.SentUnknown = time.Time{} if hand { next.SentUnknown = at.UTC() } if next.Build == k.Value.Build && next.Commit == k.Value.Commit && next.Rollout == k.Value.Rollout && next.Artifact == k.Value.Artifact && next.Package == k.Value.Package && next.Since.Equal(k.Value.Since) && next.SentUnknown.Equal(k.Value.SentUnknown) { return nil } if _, err := c.Assignments.Update(ctx, k.Key, next, k.Revision); errors.Is(err, chain.ErrMoved) { cur, found, gerr := c.Assignments.Get(ctx, k.Key) if gerr != nil { return gerr } if !found || !cur.Value.Stands() { return nil } k = chain.Kept[chain.Assignment]{Key: k.Key, Value: cur.Value, Revision: cur.Revision} continue } else { return err } } return fmt.Errorf("the assignment kept moving under the write") } func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, commit string, rollout rolloutOf, at time.Time) error { key := assignmentKey(node, module) var build, rolloutID string if sent, ours := rollout.modules[module]; ours && sent.Commit == commit { build, rolloutID = sent.Build, rollout.id } var published *chain.Published resolved := false for try := 0; try < casTries; try++ { cur, found, err := c.Assignments.Get(ctx, key) if err != nil { return err } if found && cur.Value.State != "" { return nil // a wait for a build: the module is not on the machine by it } standing := found && cur.Value.Stands() same := standing && cur.Value.Commit == commit if same && cur.Value.Build != "" && cur.Value.SentUnknown.IsZero() && (rolloutID == "" || cur.Value.Rollout != "") { return nil // a send that carries what the machine runs already says nothing new } if !resolved { // The build of the commit is looked up once, and only for a module that moved or names no build. if build == "" { if build, err = i.BuildOf(ctx, module, commit); err != nil { return err } } if build != "" { if b, found, err := i.BuildByID(ctx, build); err != nil { return err } else if found { published = publishedOf(b.Made) } } resolved = true } next := cur.Value switch { case !standing: // Carried and assigned to nobody here: a standing key, no time assigned, marked pulled in. An ended one is // written over, the ended one staying in history. next = chain.Assignment{Node: node, Module: module, PulledIn: true, Commit: commit, Build: build, Rollout: rolloutID, Since: at.UTC()} case same: // Already running it: fill what is missing, blank nothing. if next.Build == "" { next.Build = build } if next.Rollout == "" { next.Rollout = rolloutID } if next.Since.IsZero() { next.Since = at.UTC() } default: next.Commit, next.Build, next.Rollout, next.Since = commit, build, rolloutID, at.UTC() next.Artifact, next.Package = "", nil } next.SentUnknown = time.Time{} if published != nil && next.Artifact == "" && next.Package == nil && next.Build == build { next.Artifact, next.Package = published.Artifact, published.Package } if standing && next.Build == cur.Value.Build && next.Commit == cur.Value.Commit && next.Rollout == cur.Value.Rollout && next.Artifact == cur.Value.Artifact && (next.Package == nil) == (cur.Value.Package == nil) && next.Since.Equal(cur.Value.Since) && next.SentUnknown.Equal(cur.Value.SentUnknown) { return nil } if !found { if _, err := c.Assignments.Create(ctx, key, next); errors.Is(err, chain.ErrExists) { continue } else { return err } } if _, err := c.Assignments.Update(ctx, key, next, cur.Revision); errors.Is(err, chain.ErrMoved) { continue } else { return err } } return fmt.Errorf("the assignment kept moving under the write") }