mesh-controller/internal/inventory/chain_move.go

package inventory

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/chain"
)

// **The move** (novox/hq ADR 0408 §8, issue 981 note 7 Q2): the serving controller writes the rows of a kind into its
// bucket at its start, holding the lease and after the buckets exist, and records the kind in chain_moved. Idempotent: a
// key that exists is already moved and is never overwritten, so a bucket value written since is not laid over by the row
// it came from. It counts the rows against the keys, logs the largest value, and **refuses** (a Refusal, which the
// controller raises as an urgent condition) on a count that differs or a value over the bus's largest, with nothing
// switched: a kind is recorded moved only once all of it is in the bucket.

// Refusal is a move that stopped without switching a kind, and why. It is permanent until the cause is put right; any
// other error of a move (the bus away, a write not granted) is tried again.
type Refusal struct {
	Kind, Why string
}

func (r *Refusal) Error() string {
	return fmt.Sprintf("the move of %s to the bus is refused: %s", r.Kind, r.Why)
}

// MoveReport is what a move of one kind did.
type MoveReport struct {
	Kind string
	// Rows is how many records the tables held, Written how many keys this move made, and Present how many already
	// held a value (moved by an earlier start).
	Rows, Written, Present int
	// TableRows is how many rows of the tables those records came from: a request and its outcome are two rows and one
	// record.
	TableRows int
	// Largest is the biggest value, in bytes.
	Largest int
	// Already is a kind an earlier start moved: nothing was done.
	Already bool
}

// mover is one kind of the chain: its rows, read from its tables, as the keys and values they become.
type mover struct {
	kind string
	// rows reads the tables: each record's key and its value, whole, and how many rows of the tables they came from.
	rows func(ctx context.Context) ([]moveRow, int, error)
	// create writes one record under its key; chain.ErrExists when the key holds a value.
	create func(ctx context.Context, c *chain.Chain, row moveRow) error
	// keys are the keys the bucket holds.
	keys func(ctx context.Context, c *chain.Chain) ([]string, error)
	// count is how many rows the tables hold now, counted as rows reports them.
	count func(ctx context.Context, tx pgx.Tx) (int, error)
	// lock keeps the tables from taking a row while the switch is recorded.
	lock string
}

type moveRow struct {
	key   string
	value any
	size  int
}

func (i *Inventory) movers() []mover {
	return []mover{{
		kind: KindBuilds,
		rows: func(ctx context.Context) ([]moveRow, int, error) {
			builds, err := i.allBuildRows(ctx)
			if err != nil {
				return nil, 0, err
			}
			requests, err := i.allRequestRows(ctx)
			if err != nil {
				return nil, 0, err
			}
			// A build request is the build's record in state asked, and the request stays beside the outcome that
			// answered it: one record under one id (issue 1091).
			byID := make(map[string]*chain.Build, len(builds))
			var order []string
			for _, b := range builds {
				v := toChainBuild(b)
				byID[b.ID] = &v
				order = append(order, b.ID)
			}
			for _, r := range requests {
				if v, answered := byID[r.ID]; answered {
					v.Request = toRequest(r)
					continue
				}
				v := askedRecord(r)
				if r.NotAsked != "" {
					// Never handed over: asked again once, after the move (askAgain).
					v.Request.Reask = chain.ReaskPending
				}
				byID[r.ID] = &v
				order = append(order, r.ID)
			}
			out := make([]moveRow, 0, len(order))
			for _, id := range order {
				body, err := json.Marshal(byID[id])
				if err != nil {
					return nil, 0, err
				}
				out = append(out, moveRow{key: chainBuildKey(id), value: *byID[id], size: len(body)})
			}
			return out, len(builds) + len(requests), nil
		},
		create: func(ctx context.Context, c *chain.Chain, row moveRow) error {
			_, err := c.Builds.Create(ctx, row.key, row.value.(chain.Build))
			return err
		},
		keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Builds.Keys(ctx) },
		count: func(ctx context.Context, tx pgx.Tx) (int, error) {
			var n int
			err := tx.QueryRow(ctx, `select (select count(*) from build) + (select count(*) from build_request)`).Scan(&n)
			return n, err
		},
		lock: `lock table build, build_request in share mode`,
	}, {
		kind: KindRollouts,
		rows: func(ctx context.Context) ([]moveRow, int, error) {
			plans, err := i.plans(ctx, `order by created`)
			if err != nil {
				return nil, 0, err
			}
			merges, err := i.merges(ctx, ``)
			if err != nil {
				return nil, 0, err
			}
			tablePlans := len(plans)
			if plans, err = holdMerges(plans, merges); err != nil {
				return nil, 0, err
			}
			out := make([]moveRow, 0, len(plans))
			for _, p := range plans {
				body, err := json.Marshal(p)
				if err != nil {
					return nil, 0, err
				}
				out = append(out, moveRow{key: p.ID, value: p, size: len(body)})
			}
			return out, tablePlans + len(merges), nil
		},
		create: func(ctx context.Context, c *chain.Chain, row moveRow) error {
			_, err := chain.RolloutsOf[Plan](c).Create(ctx, row.key, row.value.(Plan))
			return err
		},
		keys: func(ctx context.Context, c *chain.Chain) ([]string, error) {
			return chain.RolloutsOf[Plan](c).Keys(ctx)
		},
		count: func(ctx context.Context, tx pgx.Tx) (int, error) {
			var n int
			err := tx.QueryRow(ctx, `select (select count(*) from release_plan) + (select count(*) from batched_merge)`).Scan(&n)
			return n, err
		},
		lock: `lock table release_plan, batched_merge in share mode`,
	}}
}

// holdMerges puts each batched merge into the record of the plan that answers it. A merge no plan answers — the plan_id
// of a merge a failed walk carried is null, or names a plan the store never kept — is held, flagged waiting, by the
// newest plan that lists it among its merges, else the record kept for the merges no plan answers: the record that held it
// last.
func holdMerges(plans []Plan, merges []BatchedMerge) ([]Plan, error) {
	waitingAt := -1
	at := make(map[string]int, len(plans))
	for k, p := range plans {
		at[p.ID] = k
	}
	newest := func(match func(Plan) bool) int {
		best := -1
		for k, p := range plans {
			if match(p) && (best < 0 || p.Created.After(plans[best].Created)) {
				best = k
			}
		}
		return best
	}
	for _, m := range merges {
		held := heldOf(m, m.Heard)
		holder, answered := at[m.Plan]
		if m.Plan == "" || !answered {
			held.Waiting = true
			holder = newest(func(p Plan) bool {
				if p.Delivery == nil {
					return false
				}
				for _, listed := range p.Delivery.Merges {
					if strings.EqualFold(listed.Repository, m.Repository) && listed.Commit == m.Commit {
						return true
					}
				}
				return false
			})
			if holder < 0 {
				// Held by the record kept for the merges no plan answers, made with the others.
				if waitingAt < 0 {
					plans = append(plans, Plan{ID: WaitingRecord, State: "merges", Created: time.Unix(0, 0).UTC(), Revision: 1,
						Tiers: [][]string{}, Modules: map[string]*PlanModule{}, Note: "the merges no plan answers"})
					waitingAt = len(plans) - 1
				}
				holder = waitingAt
			}
		}
		plans[holder].Held = append(plans[holder].Held, held)
	}
	return plans, nil
}

// MoveChain moves every kind not moved yet, in order, and says what it did. It needs the chain attached and the lease
// held (the chain's writes ask). It stops at the first kind that fails.
func (i *Inventory) MoveChain(ctx context.Context, say func(format string, args ...any)) ([]MoveReport, error) {
	if say == nil {
		say = func(string, ...any) {}
	}
	if i.attached() == nil {
		return nil, ErrNoChain
	}
	var out []MoveReport
	for _, m := range i.movers() {
		r, err := i.moveKind(ctx, m, say)
		if err != nil {
			return out, err
		}
		out = append(out, r)
	}
	return out, nil
}

// movePasses is how often a move reads the tables again when they took a row while it wrote: the controller before this
// one may still have been writing as the lease changed hands.
const movePasses = 5

func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...any)) (MoveReport, error) {
	report := MoveReport{Kind: m.kind}
	moved, err := i.ChainMoved(ctx)
	if err != nil {
		return report, err
	}
	if _, done := moved[m.kind]; done {
		report.Already = true
		return report, nil
	}
	for pass := 1; pass <= movePasses; pass++ {
		rows, tableRows, err := m.rows(ctx)
		if err != nil {
			return report, fmt.Errorf("the %s cannot be read from the tables: %w", m.kind, err)
		}
		report.Rows, report.TableRows, report.Largest = len(rows), tableRows, 0
		// Every value measured before any is written: a value over the bus's largest stops the move with nothing
		// written (ADR 0408 §6), never cut.
		for _, r := range rows {
			report.Largest = max(report.Largest, r.size)
			if r.size > broker.ChainValueBytes {
				return report, &Refusal{m.kind, fmt.Sprintf("the value of %s is %d bytes and the bus holds at most %d; "+
					"nothing is cut, so the record must be made smaller first", r.key, r.size, broker.ChainValueBytes)}
			}
		}
		report.Written, report.Present = 0, 0
		for _, r := range rows {
			switch err := m.create(ctx, i.attached(), r); {
			case err == nil:
				report.Written++
			case errors.Is(err, chain.ErrExists):
				report.Present++
			case errors.Is(err, chain.ErrBadKey):
				return report, &Refusal{m.kind, fmt.Sprintf("%s cannot be a key of the bus: %v", r.key, err)}
			default:
				return report, fmt.Errorf("%s %s is not written to the bus: %w", m.kind, r.key, err)
			}
		}
		// Rows against keys: every row's key is in the bucket.
		keys, err := m.keys(ctx, i.attached())
		if err != nil {
			return report, fmt.Errorf("the keys of the %s cannot be read back: %w", m.kind, err)
		}
		have := make(map[string]bool, len(keys))
		for _, k := range keys {
			have[k] = true
		}
		for _, r := range rows {
			if !have[r.key] {
				return report, &Refusal{m.kind, fmt.Sprintf("the tables hold %d records and the bucket does not hold %s", len(rows), r.key)}
			}
		}
		settled, err := i.switchKind(ctx, m, report)
		if err != nil {
			return report, err
		}
		if settled {
			say("moved the %s to the bus: %d records, %d written now and %d already there, the largest %d bytes",
				m.kind, report.Rows, report.Written, report.Present, report.Largest)
			return report, nil
		}
		say("the %s took a row while they were moved, so they are read again (pass %d of %d)", m.kind, pass, movePasses)
	}
	return report, &Refusal{m.kind, fmt.Sprintf("the tables kept taking rows through %d passes; another controller is still writing them", movePasses)}
}

// switchKind records the kind as moved, in one transaction that holds the tables against writes and finds their count
// as the move read it: false when they took a row since.
func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool, error) {
	tx, err := i.store.Pool().Begin(ctx)
	if err != nil {
		return false, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if _, err := tx.Exec(ctx, m.lock); err != nil {
		return false, err
	}
	n, err := m.count(ctx, tx)
	if err != nil {
		return false, err
	}
	if n != r.TableRows {
		return false, nil
	}
	counts, _ := json.Marshal(map[string]int{m.kind: r.Rows})
	largest, _ := json.Marshal(map[string]int{m.kind: r.Largest})
	if _, err := tx.Exec(ctx,
		`insert into chain_moved (counts, largest) values ($1, $2)
		 on conflict (only_one) do update
		    set counts = chain_moved.counts || excluded.counts, largest = chain_moved.largest || excluded.largest,
		        moved_at = now()`, counts, largest); err != nil {
		return false, err
	}
	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	i.moved.mu.Lock()
	if i.moved.kind == nil {
		i.moved.kind = map[string]bool{}
	}
	i.moved.kind[m.kind] = true
	i.moved.mu.Unlock()
	return true, nil
}

// allBuildRows is every build of the table, whole.
func (i *Inventory) allBuildRows(ctx context.Context) ([]Build, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select id, repository, ref, coalesce(module, ''), commit_hash, built_on, failed, made, source_path, manifest,
		        built_against, built_contexts, build_sources, source_fingerprint, asked, at
		   from build order by at, id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Build
	for rows.Next() {
		var b Build
		var made, manifest, against, read, sources []byte
		var asked *time.Time
		if err := rows.Scan(&b.ID, &b.Repository, &b.Ref, &b.Module, &b.Commit, &b.On, &b.Failed, &made, &b.Path,
			&manifest, &against, &read, &sources, &b.SourceFingerprint, &asked, &b.At); err != nil {
			return nil, err
		}
		if asked != nil {
			b.Asked = *asked
		}
		b.Manifest = manifest
		for _, part := range []struct {
			raw []byte
			to  any
			who string
		}{{made, &b.Made, "made"}, {against, &b.Against, "built_against"}, {read, &b.Read, "built_contexts"},
			{sources, &b.Sources, "build_sources"}} {
			if len(part.raw) == 0 {
				continue
			}
			if err := json.Unmarshal(part.raw, part.to); err != nil {
				return nil, fmt.Errorf("build %s: %s is not readable: %w", b.ID, part.who, err)
			}
		}
		out = append(out, b)
	}
	return out, rows.Err()
}

// allRequestRows is every build request of the table.
func (i *Inventory) allRequestRows(ctx context.Context) ([]BuildRequest, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select id, repository, seat, source_path, ref, commit_hash, asked_for, coalesce(not_asked, ''),
		        coalesce(outcome_unknown, ''), asked_at, at_terminal
		   from build_request order by asked_at, id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []BuildRequest
	for rows.Next() {
		var a BuildRequest
		if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked,
			&a.OutcomeUnknown, &a.At, &a.AtTerminal); err != nil {
			return nil, err
		}
		out = append(out, a)
	}
	return out, rows.Err()
}