mesh-catalog/modules/mesh-issues/cmd/mesh-issues/changes.go

package main

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

// The tracker's record is its log `changes`, not its bucket (novox/hq ADR 0300 §1). Every write is one
// operation appended under the issue's number, refused when another operation on that issue came in between;
// only then is the folded issue written to the bucket `issues`, which is what the operations add up to (§3).
const ChangesLog = "changes"

// Log is the module's log as the runtime offers it (novox/hq ADR 0297): an append at the key's last sequence,
// refused with stdio.ErrChanged when another entry came in between, and a read in pages.
type Log interface {
	AppendAt(key string, value any, expect uint64) (uint64, error)
	Read(from uint64, keys string, limit int) (*stdio.LogPage, error)
}

// runtimeLog is Log through the runtime that launched this bundle.
type runtimeLog struct{}

func (runtimeLog) AppendAt(key string, value any, expect uint64) (uint64, error) {
	return stdio.Log(ChangesLog).AppendAt(key, value, expect)
}
func (runtimeLog) Read(from uint64, keys string, limit int) (*stdio.LogPage, error) {
	return stdio.Log(ChangesLog).Read(from, keys, limit)
}

// errNoLog is a tracker started without its log: the terminal's dry run, which writes nothing.
var errNoLog = errors.New("this tracker runs without its log, so it writes nothing")

// logPage is how many entries one read of the log asks for: the most the runtime gives.
const logPage = 1000

// The kinds of operation. An operation that begins an issue holds it whole: opened, imported, carried.
const (
	OpOpened   = "opened"
	OpImported = "imported"
	// OpCarried is a value the bucket kept when the tracker switched to its log (ADR 0300 §6), whole.
	OpCarried = "carried"
)

// Operation is one change to one issue, as the log keeps it: what changed, who acted and when (ADR 0300 §1).
// One that begins an issue, or carries a value the bucket kept, holds the issue whole; every other one names
// the fields it gave a new value, the lists it added to at their end, and the fields it took away. The fold
// applies them as they are written, so it needs no knowledge of the kind: a new kind is folded as it comes.
type Operation struct {
	Op    string                     `json:"op"`
	By    string                     `json:"by"`
	At    time.Time                  `json:"at"`
	Issue *Issue                     `json:"issue,omitempty"`
	Set   map[string]json.RawMessage `json:"set,omitempty"`
	Add   map[string]json.RawMessage `json:"add,omitempty"`
	Unset []string                   `json:"unset,omitempty"`
	// Revision is the bucket's revision of the value a carried operation holds: how a carry knows which kept
	// values the log holds already.
	Revision uint64 `json:"revision,omitempty"`
}

// Act is who changes an issue and as what kind of operation.
type Act struct {
	Op string
	By string
}

// fields is an issue as its fields in JSON, without the sequence it was folded at: that is the log's, not
// the issue's.
func fields(is *Issue) (map[string]json.RawMessage, error) {
	raw, err := json.Marshal(is)
	if err != nil {
		return nil, err
	}
	var m map[string]json.RawMessage
	if err := json.Unmarshal(raw, &m); err != nil {
		return nil, err
	}
	delete(m, "folded")
	return m, nil
}

// operation is what changed from before to after as one operation of act, and false when nothing did.
func operation(act Act, at time.Time, before, after *Issue) (*Operation, bool, error) {
	b, err := fields(before)
	if err != nil {
		return nil, false, err
	}
	a, err := fields(after)
	if err != nil {
		return nil, false, err
	}
	op := &Operation{Op: act.Op, By: act.By, At: at}
	for k, v := range a {
		old, had := b[k]
		if had && bytes.Equal(old, v) {
			continue
		}
		if !had {
			// A list left out while empty: its first element is added like any other.
			old = json.RawMessage("[]")
		}
		if tail, ok := appended(old, v); ok {
			if op.Add == nil {
				op.Add = map[string]json.RawMessage{}
			}
			op.Add[k] = tail
			continue
		}
		if op.Set == nil {
			op.Set = map[string]json.RawMessage{}
		}
		op.Set[k] = v
	}
	for k := range b {
		if _, still := a[k]; !still {
			op.Unset = append(op.Unset, k)
		}
	}
	sort.Strings(op.Unset)
	return op, len(op.Set)+len(op.Add)+len(op.Unset) > 0, nil
}

// appended is what a list gained at its end, when after is before with elements added and nothing else
// changed: a note or a link is kept in the log as itself, not as every note again.
func appended(before, after json.RawMessage) (json.RawMessage, bool) {
	var b, a []json.RawMessage
	if len(after) == 0 || after[0] != '[' || json.Unmarshal(before, &b) != nil || json.Unmarshal(after, &a) != nil || len(a) <= len(b) {
		return nil, false
	}
	for i := range b {
		if !bytes.Equal(compact(b[i]), compact(a[i])) {
			return nil, false
		}
	}
	tail, err := json.Marshal(a[len(b):])
	if err != nil {
		return nil, false
	}
	return tail, true
}

func compact(raw json.RawMessage) []byte {
	var buf bytes.Buffer
	if json.Compact(&buf, raw) != nil {
		return raw
	}
	return buf.Bytes()
}

// ErrNoBeginning is an operation on an issue the log holds no beginning for.
var ErrNoBeginning = errors.New("the log holds no operation that begins it")

// fold applies one operation to an issue (nil before its first) and answers the issue it adds up to, at seq.
func fold(base *Issue, op *Operation, seq uint64) (*Issue, error) {
	var m map[string]json.RawMessage
	switch {
	case op.Issue != nil:
		f, err := fields(op.Issue)
		if err != nil {
			return nil, err
		}
		m = f
	case base == nil:
		return nil, fmt.Errorf("a %s operation at %d: %w", op.Op, seq, ErrNoBeginning)
	default:
		f, err := fields(base)
		if err != nil {
			return nil, err
		}
		m = f
	}
	for k, v := range op.Set {
		m[k] = v
	}
	for k, tail := range op.Add {
		var have, more []json.RawMessage
		if old, ok := m[k]; ok {
			if err := json.Unmarshal(old, &have); err != nil {
				return nil, fmt.Errorf("a %s operation at %d adds to %s, which is not a list", op.Op, seq, k)
			}
		}
		if err := json.Unmarshal(tail, &more); err != nil {
			return nil, fmt.Errorf("a %s operation at %d adds to %s what is not a list", op.Op, seq, k)
		}
		joined, err := json.Marshal(append(have, more...))
		if err != nil {
			return nil, err
		}
		m[k] = joined
	}
	for _, k := range op.Unset {
		delete(m, k)
	}
	raw, err := json.Marshal(m)
	if err != nil {
		return nil, err
	}
	var is Issue
	if err := json.Unmarshal(raw, &is); err != nil {
		return nil, fmt.Errorf("a %s operation at %d folds to no issue: %v", op.Op, seq, err)
	}
	is.Folded = seq
	return &is, nil
}

// foldEntry is fold of one entry of the log.
func foldEntry(base *Issue, e stdio.LogEntry) (*Issue, error) {
	var op Operation
	if err := json.Unmarshal(e.Value, &op); err != nil {
		return nil, fmt.Errorf("the entry at %d is no operation this module reads: %v", e.Sequence, err)
	}
	return fold(base, &op, e.Sequence)
}

// readKey is every entry of one issue's key from the sequence from on, oldest first, read page by page.
func (t *Tracker) readKey(n uint64, from uint64) ([]stdio.LogEntry, error) {
	if t.Log == nil {
		return nil, errNoLog
	}
	var out []stdio.LogEntry
	for from != 0 || out == nil {
		p, err := t.Log.Read(max(from, 1), keyOf(n), logPage)
		if err != nil {
			return nil, fmt.Errorf("the log of issue %d could not be read: %v", n, err)
		}
		if out == nil {
			out = []stdio.LogEntry{}
		}
		out = append(out, p.Entries...)
		from = p.Next
	}
	return out, nil
}

// append adds one operation to an issue at the sequence its fold was read at. A refusal is stdio.ErrChanged:
// another operation on the issue came in between, and nothing was written.
func (t *Tracker) appendOp(n uint64, op *Operation, expect uint64) (uint64, error) {
	if t.Log == nil {
		return 0, errNoLog
	}
	raw, err := json.Marshal(op)
	if err != nil {
		return 0, err
	}
	return t.Log.AppendAt(keyOf(n), json.RawMessage(raw), expect)
}

// errBegun is a first operation refused because the log already begins that issue, with another one.
var errBegun = errors.New("the log already holds another issue under that number")

// begin appends the first operation of a new issue — expect 0 — and answers its sequence. A first operation
// whose answer was lost is known as its own by the issue's opening id. Any other refusal is errBegun; a
// failure is answered as it is (ADR 0300 §2: the number it was given stays spent).
func (t *Tracker) begin(is *Issue, act Act) (uint64, error) {
	op := &Operation{Op: act.Op, By: act.By, At: t.now(), Issue: is}
	var err error
	for i := 0; i < createAttempts; i++ {
		if i > 0 {
			t.Sleep(jitter(time.Duration(i) * 50 * time.Millisecond))
		}
		var seq uint64
		seq, err = t.appendOp(is.Number, op, 0)
		if err == nil {
			return seq, nil
		}
		if errors.Is(err, stdio.ErrChanged) {
			first, lerr := t.firstOf(is.Number)
			if lerr != nil {
				return 0, fmt.Errorf("the log refused issue %d's first operation and could not be read back (%v): %w",
					is.Number, lerr, errBegun)
			}
			if mine(first, is.Opening) {
				return first.Sequence, nil
			}
			return 0, errBegun
		}
	}
	// An append that timed out may still have landed: read back before saying it did not.
	first, lerr := t.firstOf(is.Number)
	switch {
	case lerr != nil:
		return 0, fmt.Errorf("%v, and the log could not be read back (%v): %w", err, lerr, errUnsure)
	case mine(first, is.Opening):
		return first.Sequence, nil
	case first != nil:
		return 0, errBegun
	}
	return 0, err
}

// firstOf is an issue's first entry in the log, the one that begins it, or nil when it has none. The first, not
// the last: an open whose answer was lost may have been followed by another operation on it already, such as the
// triage's.
func (t *Tracker) firstOf(n uint64) (*stdio.LogEntry, error) {
	p, err := t.Log.Read(1, keyOf(n), 1)
	if err != nil {
		return nil, err
	}
	if len(p.Entries) == 0 {
		return nil, nil
	}
	return &p.Entries[0], nil
}

// errUnsure is a first operation the log may have taken: its answer was lost, and the log could not be read back.
var errUnsure = errors.New("it may have been taken")

// mine says whether a log entry is the first operation of the issue opened by opening.
func mine(e *stdio.LogEntry, opening string) bool {
	if e == nil || opening == "" {
		return false
	}
	var op Operation
	return json.Unmarshal(e.Value, &op) == nil && op.Issue != nil && op.Issue.Opening == opening
}

// change reads an issue, applies f, appends what changed to the log at the sequence the issue was folded at,
// and then writes the fold to the bucket (ADR 0300 §1). An append another operation won is read and applied
// again, so no note, link or move is lost and none is applied to a stale issue; a refused append writes
// nothing. A change that changes nothing appends nothing.
func (t *Tracker) change(n uint64, act Act, f func(*Issue) error) (*Issue, error) {
	for i := 0; i < issueAttempts; i++ {
		is, rev, err := t.current(n)
		if err != nil {
			return nil, err
		}
		if is.Folded == 0 {
			// Kept before the switch, and not carried yet: carried first, so the log begins it (ADR 0300 §6).
			if _, _, err := t.carryOne(n); err != nil {
				return nil, fmt.Errorf("issue %d is kept from before the tracker's log and is not carried into it yet, "+
					"so nothing is changed: %v", n, err)
			}
			continue
		}
		before, err := clone(is)
		if err != nil {
			return nil, err
		}
		if err := f(is); err != nil {
			return nil, err
		}
		raw, err := json.Marshal(is)
		if err != nil {
			return nil, err
		}
		if len(raw) > MostIssueBytes {
			return nil, fmt.Errorf("issue %d would be %d bytes; one issue is kept in at most %d (252 KiB). Nothing "+
				"was changed: say it shorter, or in a document the issue links", n, len(raw), MostIssueBytes)
		}
		op, changed, err := operation(act, t.now(), before, is)
		if err != nil {
			return nil, err
		}
		if !changed {
			return is, nil
		}
		seq, err := t.appendOp(n, op, is.Folded)
		if errors.Is(err, stdio.ErrChanged) {
			// Another operation came in between, or the bucket is behind the log: fold what the log holds and
			// apply this change again to that.
			if _, err := t.catchUp(n); err != nil {
				return nil, err
			}
			t.Sleep(jitter(time.Duration(i+1) * 5 * time.Millisecond))
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("issue %d: the log did not take the change (%v), so the bucket was not written. If "+
				"the log took it before the failure it is kept and the bucket folds it at the next change or start: "+
				"get %d before changing it again", n, err, n)
		}
		is.Folded = seq
		t.store(n, is, rev)
		return is, nil
	}
	return nil, fmt.Errorf("issue %d was changed by another caller %d times while this change was made; nothing was "+
		"changed. Try again", n, issueAttempts)
}

// current is an issue as the bucket holds it, and the revision it was read at; an issue the log holds and the
// bucket does not is folded into the bucket first.
func (t *Tracker) current(n uint64) (*Issue, uint64, error) {
	is, rev, err := read(t.Store, n)
	if errors.Is(err, ErrNoIssue) && t.Log != nil {
		if _, ferr := t.catchUp(n); ferr != nil {
			return nil, 0, ferr
		}
		is, rev, err = read(t.Store, n)
	}
	return is, rev, err
}

// store writes the fold of an issue at the revision read, or creates it at 0. The log already holds the
// change, so a write another writer won is folded again from the log, and one that fails is said and left for
// the next change or start, which fold the log again; it is never answered as a change not made.
func (t *Tracker) store(n uint64, is *Issue, rev uint64) {
	var err error
	if rev == 0 {
		_, err = t.Store.Create(IssuesState, keyOf(n), is)
	} else {
		_, err = t.Store.Update(IssuesState, keyOf(n), is, rev)
	}
	if err == nil {
		t.remember(is)
		return
	}
	if errors.Is(err, stdio.ErrChanged) {
		if folded, cerr := t.catchUp(n); cerr == nil {
			if folded != nil {
				t.remember(folded)
			}
			return
		} else {
			err = cerr
		}
	}
	t.remember(is)
	t.Logf("[mesh-issues] issue %d is kept in the log at %d and its bucket could not be written (%v): the bucket is "+
		"behind until the next change of it or the next start folds it", n, is.Folded, err)
}

// catchUp folds into the bucket every operation of an issue the bucket has not folded, and answers the issue
// as folded (nil when neither holds it). The bucket is written only at the revision read; a write another
// writer won is read again.
func (t *Tracker) catchUp(n uint64) (*Issue, error) {
	for i := 0; i < issueAttempts; i++ {
		is, rev, err := read(t.Store, n)
		if err != nil && !errors.Is(err, ErrNoIssue) {
			return nil, err
		}
		var from uint64 = 1
		if is != nil {
			if is.Folded == 0 {
				// Not carried yet: what the log holds is no fold of it.
				return is, nil
			}
			from = is.Folded + 1
		}
		entries, err := t.readKey(n, from)
		if err != nil {
			return nil, err
		}
		if len(entries) == 0 {
			return is, nil
		}
		folded := is
		for _, e := range entries {
			if folded, err = foldEntry(folded, e); err != nil {
				return nil, fmt.Errorf("issue %d cannot be folded from its log: %v", n, err)
			}
		}
		if is == nil {
			_, err = t.Store.Create(IssuesState, keyOf(n), folded)
		} else {
			_, err = t.Store.Update(IssuesState, keyOf(n), folded, rev)
		}
		if err == nil {
			t.remember(folded)
			return folded, nil
		}
		if !errors.Is(err, stdio.ErrChanged) {
			return nil, fmt.Errorf("issue %d is behind its log in the bucket and could not be written (%v)", n, err)
		}
	}
	return nil, fmt.Errorf("issue %d was written by another writer %d times while it was folded from its log", n, issueAttempts)
}

func clone(is *Issue) (*Issue, error) {
	raw, err := json.Marshal(is)
	if err != nil {
		return nil, err
	}
	var c Issue
	return &c, json.Unmarshal(raw, &c)
}

// Carried is what the carry did (ADR 0300 §6), as it happened: the values it appended, not the ones it meant to.
type Carried struct {
	// Issues is how many issues it appended values of, Values how many values; Already how many issues the log
	// already began whole, which it left as they are.
	Issues  int `json:"issues"`
	Values  int `json:"values"`
	Already int `json:"already"`
	// Deletes is how many deletions among the kept values were passed over: an issue is never deleted by the
	// tracker, and each value after one holds the issue whole, so the fold needs none.
	Deletes int `json:"deletes,omitempty"`
	// Faults is each issue that could not be carried, and why. It stays in the bucket as it was, readable and
	// listed; every write to it is refused with the same reason until it can be carried, so nothing reaches the
	// log before its carry.
	Faults []string `json:"faults,omitempty"`
	Said   string   `json:"said"`
}

// carryOne carries one issue the bucket has not folded from the log: each value the bucket still keeps of it
// (novox/hq ADR 0307), oldest first, as `carried` operations holding the whole issue and the bucket's revision
// of that value, then the fold written to the bucket. Answers how many values it appended and how many
// deletions it passed over.
//
// It resumes rather than restarts. A value is known to the log when its revision is at or below the highest
// revision a carried operation of the issue holds, or when it is a value the tracker wrote itself (it carries
// `folded`): matched by revision, not by content, because the same content may be kept twice (a write that
// changed nothing) and a revision names one value exactly. So an interrupted carry, or another instance carrying
// the same issue at the same moment, carries each value once; a value written beside the log after the carry,
// by an instance from before the switch, is carried as one more operation and so is never overwritten.
func (t *Tracker) carryOne(n uint64) (int, int, error) {
	appended := 0
	for i := 0; i < issueAttempts; i++ {
		is, rev, err := read(t.Store, n)
		if err != nil {
			return appended, 0, err
		}
		if is.Folded != 0 {
			return appended, 0, nil
		}
		entries, err := t.readKey(n, 1)
		if err != nil {
			return appended, 0, err
		}
		var folded *Issue
		var expect, known uint64
		for _, e := range entries {
			if folded, err = foldEntry(folded, e); err != nil {
				return appended, 0, fmt.Errorf("issue %d cannot be folded from its log: %v", n, err)
			}
			expect = e.Sequence
			var op Operation
			if json.Unmarshal(e.Value, &op) == nil && op.Op == OpCarried {
				known = max(known, op.Revision)
			}
		}
		kept, err := t.Store.History(IssuesState, keyOf(n))
		if err != nil {
			return appended, 0, fmt.Errorf("the values the bucket keeps of issue %d could not be read: %v", n, err)
		}
		type value struct {
			issue    *Issue
			revision uint64
		}
		var values []value
		deletes := 0
		for _, e := range kept {
			if e.Op == "delete" {
				deletes++
				continue
			}
			var v Issue
			if err := json.Unmarshal(e.Value, &v); err != nil {
				return appended, deletes, fmt.Errorf("the value of issue %d at revision %d is in a form this module "+
					"cannot read (%v): not carried, so that nothing kept is left out", n, e.Revision, err)
			}
			switch {
			case v.Folded != 0:
				// The tracker's own fold: the log holds it already, and whatever came before it.
				values = nil
				known = max(known, e.Revision)
			case e.Revision > known:
				values = append(values, value{&v, e.Revision})
			}
		}
		if len(kept) == 0 || kept[len(kept)-1].Revision < rev {
			// The value read is newer than the history read before it: carried too.
			if rev > known {
				values = append(values, value{is, rev})
			}
		}
		raced := false
		for _, v := range values {
			v.issue.Folded = 0
			seq, err := t.appendOp(n, &Operation{Op: OpCarried, By: Seat, At: t.now(), Issue: v.issue, Revision: v.revision}, expect)
			if errors.Is(err, stdio.ErrChanged) {
				raced = true
				break
			}
			if err != nil {
				return appended, deletes, fmt.Errorf("issue %d could not be carried into the log (%v); %d of its %d "+
					"values not carried yet were carried. Nothing is lost: the bucket still holds it, and the carry "+
					"resumes after the last value carried, at the next start or before its next change", n, err, appended, len(values))
			}
			appended++
			expect = seq
			if folded, err = fold(nil, &Operation{Op: OpCarried, Issue: v.issue}, seq); err != nil {
				return appended, deletes, err
			}
		}
		if raced {
			// Another carrier was first: read again; what it appended is known by its revisions.
			continue
		}
		if folded == nil {
			return appended, deletes, fmt.Errorf("issue %d keeps no value to carry", n)
		}
		if _, err := t.Store.Update(IssuesState, keyOf(n), folded, rev); err != nil {
			if errors.Is(err, stdio.ErrChanged) {
				// Written meanwhile, by a carrier or an instance from before the switch: read again.
				continue
			}
			return appended, deletes, fmt.Errorf("issue %d was carried into the log and its bucket could not be written (%v): "+
				"it is folded at the next change or start", n, err)
		}
		t.remember(folded)
		return appended, deletes, nil
	}
	return appended, 0, fmt.Errorf("issue %d was written by another writer %d times while it was carried", n, issueAttempts)
}

// same says whether two issues hold the same, whatever sequence each was folded at.
func same(a, b *Issue) bool {
	fa, err1 := fields(a)
	fb, err2 := fields(b)
	if err1 != nil || err2 != nil || len(fa) != len(fb) {
		return false
	}
	for k, v := range fa {
		if !bytes.Equal(compact(v), compact(fb[k])) {
			return false
		}
	}
	return true
}

// Carry carries every issue the bucket holds and has not folded from the log (ADR 0300 §6). An issue that
// cannot be carried is a fault, named with its reason, and the others are carried. Run again, it carries
// nothing more, and resumes an issue whose carry was interrupted after the last value carried.
func (t *Tracker) Carry() (*Carried, error) {
	keys, err := t.Store.Keys(IssuesState)
	if err != nil {
		return nil, fmt.Errorf("the issues kept cannot be listed: %v", err)
	}
	out := &Carried{}
	for _, k := range keys {
		n, ok := numberOfKey(k)
		if !ok {
			continue
		}
		is, _, err := read(t.Store, n)
		if errors.Is(err, ErrNoIssue) {
			continue
		}
		if err != nil {
			out.Faults = append(out.Faults, err.Error())
			continue
		}
		if is.Folded != 0 {
			out.Already++
			continue
		}
		v, d, err := t.carryOne(n)
		out.Values += v
		out.Deletes += d
		if v > 0 {
			out.Issues++
		} else if err == nil {
			out.Already++
		}
		if err != nil {
			t.Logf("[mesh-issues] %v", err)
			out.Faults = append(out.Faults, err.Error())
		}
	}
	if out.Values == 0 {
		out.Said = fmt.Sprintf("Carried nothing: the log begins every issue of the bucket it could (%d).", out.Already)
	} else {
		out.Said = fmt.Sprintf("Carried %d values of %d issues into the log, each value the bucket kept, oldest first; "+
			"%d issues were there already.", out.Values, out.Issues, out.Already)
	}
	if out.Deletes > 0 {
		out.Said += fmt.Sprintf(" %d deletions among the kept values were passed over: each value after one holds "+
			"the issue whole.", out.Deletes)
	}
	if len(out.Faults) > 0 {
		out.Said += fmt.Sprintf(" %d issues could not be carried, and stay as the bucket holds them, every write to "+
			"them refused until they can be: %s.", len(out.Faults), strings.Join(out.Faults, "; "))
	}
	return out, nil
}

// Started is what the tracker did on start (ADR 0300 §3, §6).
type Started struct {
	// Entries is how many operations the log holds; Issues how many issues they add up to.
	Entries int `json:"entries"`
	Issues  int `json:"issues"`
	// Refolded is how many issues the bucket held behind the log, Rebuilt how many it did not hold at all.
	Refolded int      `json:"refolded"`
	Rebuilt  int      `json:"rebuilt"`
	Carried  *Carried `json:"carried"`
	// Counter is the issue counter after start, raised to the highest number in the log where it was below.
	Counter uint64   `json:"counter"`
	Faults  []string `json:"faults,omitempty"`
	Said    string   `json:"said"`
}

// Start carries what the bucket holds and the log does not, reads the whole log, folds every issue whose
// last operation is newer than the bucket's fold of it — an emptied bucket is rebuilt whole — keeps the
// counter at or above the highest number in the log, and holds every issue in memory with its index
// (ADR 0300 §3, §5, §6).
func (t *Tracker) Start() (*Started, error) {
	out := &Started{}
	if t.Log == nil {
		return out, errNoLog
	}
	carried, err := t.Carry()
	if err != nil {
		return out, fmt.Errorf("the carry could not start (%v); writes carry each issue before they change it", err)
	}
	out.Carried = carried
	for _, f := range carried.Faults {
		out.Faults = append(out.Faults, "not carried: "+f)
	}
	folds := map[uint64]*Issue{}
	// broken is each issue an entry of which could not be folded: none of its later entries is folded on top,
	// and its bucket value is left as it is, so no fold that skipped an operation is ever written.
	broken := map[uint64]bool{}
	var highest uint64
	for from := uint64(1); from != 0; {
		p, err := t.Log.Read(from, "", logPage)
		if err != nil {
			return out, fmt.Errorf("the log could not be read from %d: %v", from, err)
		}
		for _, e := range p.Entries {
			out.Entries++
			n, ok := numberOfKey(e.Key)
			if !ok {
				out.Faults = append(out.Faults, fmt.Sprintf("the entry at %d has the key %q, which is no issue's number", e.Sequence, e.Key))
				continue
			}
			highest = max(highest, n)
			if broken[n] {
				continue
			}
			f, err := foldEntry(folds[n], e)
			if err != nil {
				out.Faults = append(out.Faults, fmt.Sprintf("issue %d cannot be folded from its log, and its bucket "+
					"value is left as it is: %v", n, err))
				broken[n] = true
				delete(folds, n)
				continue
			}
			folds[n] = f
		}
		from = p.Next
	}
	out.Issues = len(folds) + len(broken)
	numbers := make([]uint64, 0, len(folds))
	for n := range folds {
		numbers = append(numbers, n)
	}
	sort.Slice(numbers, func(i, j int) bool { return numbers[i] < numbers[j] })
	for _, n := range numbers {
		f := folds[n]
		is, rev, err := read(t.Store, n)
		switch {
		case errors.Is(err, ErrNoIssue):
			if _, err := t.Store.Create(IssuesState, keyOf(n), f); err != nil && !errors.Is(err, stdio.ErrChanged) {
				out.Faults = append(out.Faults, fmt.Sprintf("issue %d could not be rebuilt in the bucket: %v", n, err))
				continue
			} else if err == nil {
				out.Rebuilt++
			}
		case err != nil:
			out.Faults = append(out.Faults, err.Error())
			continue
		case is.Folded == 0:
			// Written beside the log, by an instance from before the switch, after the carry, and not carried again
			// (its carry is in the faults): the fold would overwrite that write. Left for its carry, which appends it
			// as one more operation before the issue's next change.
			out.Faults = append(out.Faults, fmt.Sprintf("issue %d holds a value written beside the log and not carried "+
				"yet; it is not refolded over", n))
			continue
		case is.Folded < f.Folded:
			if _, err := t.Store.Update(IssuesState, keyOf(n), f, rev); err != nil && !errors.Is(err, stdio.ErrChanged) {
				out.Faults = append(out.Faults, fmt.Sprintf("issue %d could not be refolded in the bucket: %v", n, err))
				continue
			} else if err == nil {
				out.Refolded++
			}
		}
	}
	if out.Counter, err = t.raiseCounter(highest); err != nil {
		out.Faults = append(out.Faults, err.Error())
	}
	all, err := t.readAll()
	if err != nil {
		return out, err
	}
	t.view().load(all)
	out.Said = fmt.Sprintf("The log holds %d operations of %d issues; the bucket was behind on %d and missed %d, now "+
		"folded. %s The issue counter is at %d. %d issues are held in memory and indexed.", out.Entries, out.Issues,
		out.Refolded, out.Rebuilt, carried.Said, out.Counter, len(all))
	if len(out.Faults) > 0 {
		out.Said += fmt.Sprintf(" %d faults: %v", len(out.Faults), out.Faults)
	}
	return out, nil
}

// raiseCounter keeps the issue counter at or above the highest number in the log, so no number the log holds
// is handed out again; a counter not started yet is left to the first open, which starts it above every issue
// the bucket holds.
func (t *Tracker) raiseCounter(highest uint64) (uint64, error) {
	for i := 0; i < issueAttempts; i++ {
		e, err := t.Store.Get(CountersState, IssueCounter)
		if err != nil {
			return 0, fmt.Errorf("the issue counter cannot be read: %v", err)
		}
		if e == nil {
			return 0, nil
		}
		var n uint64
		if err := json.Unmarshal(e.Value, &n); err != nil {
			return 0, fmt.Errorf("the issue counter holds %s, no number", e.Value)
		}
		if n >= highest {
			return n, nil
		}
		_, err = t.Store.Update(CountersState, IssueCounter, highest, e.Revision)
		if err == nil {
			t.Logf("[mesh-issues] the issue counter was at %d, below issue %d in the log: raised to %d", n, highest, highest)
			return highest, nil
		}
		if !errors.Is(err, stdio.ErrChanged) {
			return n, fmt.Errorf("the issue counter is at %d, below issue %d in the log, and could not be raised: %v", n, highest, err)
		}
	}
	return 0, errors.New("the issue counter was advanced by others while it was raised to the log's highest number")
}