| 1 | package moves |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "sort" |
| 10 | "strings" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/lease" |
| 14 | "github.com/superfly/ltx" |
| 15 | ) |
| 16 | |
| 17 | // The seal (§7.3 "seal by version", §7.4 staged): the exact versions of the |
| 18 | // staging lane's replica/, files/, timers/ and state/ objects the cut wrote, the |
| 19 | // required final TXID N, and the two facts a recovery reads, in the claim's |
| 20 | // op journal under hostclaim.JournalSeal. Landing copies these versions and |
| 21 | // nothing else, so a stale round's later write (a newer version, or a |
| 22 | // delete marker) is never installed. |
| 23 | // |
| 24 | // It is written twice under one op: at the end of `stopped` with the replica |
| 25 | // and the timers (the archive does not exist yet), and at the end of |
| 26 | // `staged` with the archive's two objects added. Files is empty until then, |
| 27 | // and the cut never flips on a seal without them (HasFiles). |
| 28 | type Seal struct { |
| 29 | V int `json:"v"` |
| 30 | N uint64 `json:"n"` |
| 31 | Replica []SealedObject `json:"replica"` |
| 32 | Files []SealedObject `json:"files,omitempty"` |
| 33 | Timers []SealedObject `json:"timers,omitempty"` |
| 34 | // State is the lane's copy of the instance's build state (settings spec |
| 35 | // rev 11 §4.3a), at most one object at the lane's state/ key: the |
| 36 | // version the round that wrote this seal copied. Absent when the source |
| 37 | // had none; a seal written before this field existed decodes with none. |
| 38 | State []SealedObject `json:"state,omitempty"` |
| 39 | // SrcCovered is whether the source's own replica of holder.generation |
| 40 | // reached the stopped database's position with no pending write (§7.4, |
| 41 | // rev 7.11): it decides a rollback's clean return (§7.8). |
| 42 | SrcCovered bool `json:"src_covered"` |
| 43 | // FromDB is whether the seal was built from the stopped database itself |
| 44 | // (§5.1 step 9): false for a source that had no local database. |
| 45 | FromDB bool `json:"from_db"` |
| 46 | SealedAt time.Time `json:"sealed_at"` |
| 47 | // LeaseUntil is the latest moment a request signed by a staging writer |
| 48 | // of this lane can still land: the last guarded signature's time plus |
| 49 | // SignatureLife plus the skew (ruling G7). Bound returns it. |
| 50 | LeaseUntil time.Time `json:"lease_until"` |
| 51 | } |
| 52 | |
| 53 | // SealVersion is the only seal format this code reads or writes. |
| 54 | const SealVersion = 1 |
| 55 | |
| 56 | // ErrBadSeal is a seal that pins nothing usable: refused on write and on |
| 57 | // read, never repaired. |
| 58 | var ErrBadSeal = errors.New("moves: the seal is not usable") |
| 59 | |
| 60 | // Bound is the time after which no staging writer that held the lease can |
| 61 | // still write: the staging park's evidence_at (Plan 5). |
| 62 | func (s Seal) Bound() time.Time { return s.LeaseUntil } |
| 63 | |
| 64 | // HasFiles reports whether the seal names the archive: `archived` and the |
| 65 | // flip require it. |
| 66 | func (s Seal) HasFiles() bool { return len(s.Files) > 0 } |
| 67 | |
| 68 | // ReachesN reports whether the sealed replica ends exactly at N: its newest |
| 69 | // level file's max TXID is N. A set that stops short cannot restore to N, |
| 70 | // and one that runs past it is not this stopped database's. |
| 71 | func (s Seal) ReachesN() bool { |
| 72 | var newest ltx.TXID |
| 73 | for _, o := range s.Replica { |
| 74 | pos, err := sealedPosition(o.Key) |
| 75 | if err != nil { |
| 76 | return false |
| 77 | } |
| 78 | newest = max(newest, pos.max) |
| 79 | } |
| 80 | return s.N > 0 && uint64(newest) == s.N |
| 81 | } |
| 82 | |
| 83 | // Validate refuses a seal that pins nothing: another version, no position, |
| 84 | // no replica, any object without a key or a usable version id (G27: "" and |
| 85 | // "null" pin nothing), a key named twice, a negative size, a replica object |
| 86 | // that is not a level file or is below an LTX header (Task 8's |
| 87 | // NewSealedClient refuses those), no seal time, and no bound or a bound |
| 88 | // before the seal. |
| 89 | func (s Seal) Validate() error { |
| 90 | if s.V != SealVersion { |
| 91 | return fmt.Errorf("%w: version %d", ErrBadSeal, s.V) |
| 92 | } |
| 93 | if s.N == 0 || len(s.Replica) == 0 { |
| 94 | return fmt.Errorf("%w: no position or no replica", ErrBadSeal) |
| 95 | } |
| 96 | if s.SealedAt.IsZero() || s.LeaseUntil.IsZero() || s.LeaseUntil.Before(s.SealedAt) { |
| 97 | return fmt.Errorf("%w: no seal time or no writers' bound", ErrBadSeal) |
| 98 | } |
| 99 | seen := map[string]bool{} |
| 100 | if len(s.State) > 1 || (len(s.State) == 1 && !strings.HasSuffix(s.State[0].Key, BuildStateSuffix)) { |
| 101 | return fmt.Errorf("%w: the state class names %d objects, want one at %s", ErrBadSeal, len(s.State), BuildStateSuffix) |
| 102 | } |
| 103 | for _, class := range [][]SealedObject{s.Replica, s.Files, s.Timers, s.State} { |
| 104 | for _, o := range class { |
| 105 | if o.Key == "" || !lease.UsableVersionID(o.VersionID) || o.Size < 0 || seen[o.Key] { |
| 106 | return fmt.Errorf("%w: object %q version %q size %d", ErrBadSeal, o.Key, o.VersionID, o.Size) |
| 107 | } |
| 108 | seen[o.Key] = true |
| 109 | } |
| 110 | } |
| 111 | for _, o := range s.Replica { |
| 112 | if _, err := sealedPosition(o.Key); err != nil { |
| 113 | return fmt.Errorf("%w: %v", ErrBadSeal, err) |
| 114 | } |
| 115 | if o.Size < ltx.HeaderSize { |
| 116 | return fmt.Errorf("%w: replica object %s is below an LTX header", ErrBadSeal, o.Key) |
| 117 | } |
| 118 | } |
| 119 | return nil |
| 120 | } |
| 121 | |
| 122 | func sortObjects(objs []SealedObject) []SealedObject { |
| 123 | out := append([]SealedObject(nil), objs...) |
| 124 | sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key }) |
| 125 | return out |
| 126 | } |
| 127 | |
| 128 | // Encode is the seal as the claim journal carries it: JSON with every |
| 129 | // object list sorted by key, so one seal always encodes to one string. |
| 130 | func (s Seal) Encode() string { |
| 131 | c := s |
| 132 | c.Replica, c.Files, c.Timers, c.State = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers), sortObjects(s.State) |
| 133 | c.SealedAt, c.LeaseUntil = s.SealedAt.UTC(), s.LeaseUntil.UTC() |
| 134 | b, err := json.Marshal(c) |
| 135 | if err != nil { |
| 136 | // Every field is a plain value: Marshal cannot fail. |
| 137 | panic(err) |
| 138 | } |
| 139 | return string(b) |
| 140 | } |
| 141 | |
| 142 | // DecodeSeal reads a journalled seal strictly (no unknown field, no |
| 143 | // trailing data) and validates it. Anything else is ErrBadSeal. |
| 144 | func DecodeSeal(raw string) (Seal, error) { |
| 145 | dec := json.NewDecoder(bytes.NewReader([]byte(raw))) |
| 146 | dec.DisallowUnknownFields() |
| 147 | var s Seal |
| 148 | if err := dec.Decode(&s); err != nil { |
| 149 | return Seal{}, fmt.Errorf("%w: %v", ErrBadSeal, err) |
| 150 | } |
| 151 | if _, err := dec.Token(); err != io.EOF { |
| 152 | return Seal{}, fmt.Errorf("%w: trailing data", ErrBadSeal) |
| 153 | } |
| 154 | if err := s.Validate(); err != nil { |
| 155 | return Seal{}, err |
| 156 | } |
| 157 | s.Replica, s.Files, s.Timers, s.State = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers), sortObjects(s.State) |
| 158 | return s, nil |
| 159 | } |
| 160 | |
amadan