| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "database/sql" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/hostguard" |
| 14 | "github.com/carlosframework/platform/internal/moves" |
| 15 | "github.com/carlosframework/platform/internal/park" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | "strconv" |
| 18 | ) |
| 19 | |
| 20 | func claimRecoveryObservation(o hostclaim.Observed) bool { |
| 21 | return o.Claim != nil && claimRecoveryKind(o.Claim.Op.Kind) |
| 22 | } |
| 23 | func recoveryPredecessor(o hostclaim.Observed) hostclaim.AdmissionRecord { |
| 24 | h := o.Claim.Holder |
| 25 | p := hostclaim.AdmissionRecord{Box: h.WriterBox, Session: h.Session, Generation: h.Generation} |
| 26 | if len(h.Admissions) > 0 { |
| 27 | p = h.Admissions[len(h.Admissions)-1] |
| 28 | } |
| 29 | return p |
| 30 | } |
| 31 | func (h claimCrossBoxHandover) recoveryEvidence(ctx context.Context, o hostclaim.Observed) error { |
| 32 | if !claimRecoveryObservation(o) || o.Absent || o.ETag == "" || o.Claim.Holder == nil || h.Evidence == nil || h.Box == "" { |
| 33 | return hostbox.ErrChanged |
| 34 | } |
| 35 | c := o.Claim |
| 36 | p := recoveryPredecessor(o) |
| 37 | op := c.Op |
| 38 | if claimInitialSourceRecovery(o, h.Box) { |
| 39 | return h.sourceRecoveryEvidence(ctx, o) |
| 40 | } |
| 41 | if op.ID == "" || op.Phase != hostclaim.PhaseProbing || op.Exec != h.Box || op.Intent["source"] != h.Box || p.Box == "" || p.Box == h.Box || p.Box != op.Intent["target"] || p.Box != c.Holder.WriterBox || p.Session != c.Holder.Session { |
| 42 | return hostbox.ErrChanged |
| 43 | } |
| 44 | lossID, err := hostclaim.OperationLossID(c) |
| 45 | if err != nil { |
| 46 | return err |
| 47 | } |
| 48 | if lossID != "" { |
| 49 | l, err := hostclaim.ReadLossRecord(ctx, h.Evidence, h.PublicKey, lossID) |
| 50 | if err != nil { |
| 51 | return err |
| 52 | } |
| 53 | r := l.Record |
| 54 | if r.State != "committed" || r.Host != o.Host || r.OpID != op.ID || r.HolderIncarnation != c.Holder.Incarnation || r.Box != p.Box { |
| 55 | return hostclaim.ErrLossRecord |
| 56 | } |
| 57 | for _, source := range r.WriteSources { |
| 58 | if source.Reason == "target" && source.Box == p.Box && source.Session == p.Session && source.Generation == p.Generation { |
| 59 | return nil |
| 60 | } |
| 61 | } |
| 62 | return hostclaim.ErrLossRecord |
| 63 | } |
| 64 | if _, ok := hostclaim.ConsumedMoveTargetFence(c); ok { |
| 65 | return nil |
| 66 | } |
| 67 | if op.Journal["fence."+p.Box] != "" { |
| 68 | var ref hostclaim.FenceRef |
| 69 | if json.Unmarshal([]byte(op.Journal["fence."+p.Box]), &ref) != nil || op.Intent["target_enrollment"] == "" || ref.Enrollment != op.Intent["target_enrollment"] { |
| 70 | return hostclaim.ErrFence |
| 71 | } |
| 72 | verified, err := hostclaim.VerifyFence(ctx, h.Evidence, h.FenceAuthority, ref) |
| 73 | if err != nil { |
| 74 | return err |
| 75 | } |
| 76 | if (op.Intent["home_region"] != "" && op.Intent["home_region"] != verified.FleetHomeRegion()) || verified.Box() != p.Box || !park.SameRegion(op.Intent["home_region"], verified.Region(), op.Intent["to"]) { |
| 77 | return hostclaim.ErrFence |
| 78 | } |
| 79 | return nil |
| 80 | } |
| 81 | if key := op.Journal["target_stop_receipt"]; key != "" { |
| 82 | return h.recoveryStopReceipt(ctx, o, p, key, op.Journal["target_stop_sha256"]) |
| 83 | } |
| 84 | key := op.Journal[hostclaim.JournalEvidenceTarget] |
| 85 | expected := moves.StoppedKey(moves.ReportPrefix(c.Holder.Account, c.Holder.App, o.Host, op.ID), "target") |
| 86 | session, err := strconv.ParseUint(op.Journal[hostclaim.JournalTargetSession], 10, 64) |
| 87 | if err != nil || session != p.Session || op.Journal[hostclaim.JournalTargetGeneration] != p.Generation || key != expected { |
| 88 | return hostbox.ErrChanged |
| 89 | } |
| 90 | report, err := readStoppedReport(ctx, h.Evidence, key) |
| 91 | if err != nil { |
| 92 | return err |
| 93 | } |
| 94 | if report.Role != "target" || report.Box != p.Box || report.Host != o.Host || report.OpID != op.ID || report.Incarnation != c.Holder.Incarnation { |
| 95 | return hostbox.ErrChanged |
| 96 | } |
| 97 | return nil |
| 98 | } |
| 99 | func (h claimCrossBoxHandover) beforeRecovery(ctx context.Context, o hostclaim.Observed) error { |
| 100 | if err := h.recoveryEvidence(ctx, o); err != nil { |
| 101 | return err |
| 102 | } |
| 103 | old := o.Claim.Holder |
| 104 | return h.Guard.With(ctx, o.Host, hostguard.Captured{Incarnation: old.Incarnation, Stamp: old.Stamp}, func(c *hostclaim.Claim) error { |
| 105 | if !hostclaim.SameWriterIdentity(o, hostclaim.Observed{Host: o.Host, Claim: c}) { |
| 106 | return hostbox.ErrChanged |
| 107 | } |
| 108 | return nil |
| 109 | }, func() error { return nil }) |
| 110 | } |
| 111 | func (h claimCrossBoxHandover) afterRecovery(ctx context.Context, prior, admitted hostclaim.Observed) error { |
| 112 | if err := h.recoveryEvidence(ctx, prior); err != nil { |
| 113 | return err |
| 114 | } |
| 115 | if admitted.Absent || admitted.Claim == nil || admitted.Claim.Holder == nil || admitted.Host != prior.Host { |
| 116 | return hostbox.ErrChanged |
| 117 | } |
| 118 | old, next := prior.Claim.Holder, admitted.Claim.Holder |
| 119 | op, ac := prior.Claim.Op, admitted.Claim |
| 120 | if ac.Op.ID != op.ID || ac.Op.Kind != op.Kind || ac.Op.Phase != op.Phase || ac.Op.Exec != h.Box || next.Incarnation != old.Incarnation || next.Session <= old.Session || next.Stamp != next.Session || next.WriterBox != h.Box || len(next.Admissions) == 0 { |
| 121 | return hostbox.ErrChanged |
| 122 | } |
| 123 | p := recoveryPredecessor(prior) |
| 124 | a := next.Admissions[len(next.Admissions)-1] |
| 125 | if a.Session != next.Session || a.Box != h.Box || a.Mode != "cold" || a.Generation != next.Generation || a.Predecessor != p.Generation { |
| 126 | return hostbox.ErrChanged |
| 127 | } |
| 128 | lossID, err := hostclaim.OperationLossID(prior.Claim) |
| 129 | if err != nil { |
| 130 | return err |
| 131 | } |
| 132 | if lossID != "" && (a.LossID != lossID || a.Coverage != "" || (p.Session != 0 && (a.WritesLost == nil || a.WritesLost.Box != p.Box || a.WritesLost.Session != p.Session))) { |
| 133 | return hostclaim.ErrLossRecord |
| 134 | } |
| 135 | if lossID == "" && a.LossID != "" { |
| 136 | return hostclaim.ErrLossRecord |
| 137 | } |
| 138 | current, err := hostclaim.Read(ctx, h.Evidence, admitted.Host) |
| 139 | if err != nil { |
| 140 | return err |
| 141 | } |
| 142 | if !hostclaim.SameWriterIdentity(current, admitted) { |
| 143 | return hostbox.ErrChanged |
| 144 | } |
| 145 | if h.Registry == nil { |
| 146 | return hostbox.ErrTypedInventory |
| 147 | } |
| 148 | if h.Inventory.Remote() || h.Guard.FenceView != nil { |
| 149 | return h.installRecoveryRemote(ctx, prior, admitted) |
| 150 | } |
| 151 | // Activate retires the prior local stamp through the descriptor-based guard. |
| 152 | // The restore hook subsequently retires the source database and marker. |
| 153 | if err = h.Guard.Activate(ctx, admitted.Host, h.Box, nil); err != nil { |
| 154 | return err |
| 155 | } |
| 156 | return h.Guard.With(ctx, admitted.Host, hostguard.Captured{Incarnation: next.Incarnation, Stamp: next.Stamp}, func(c *hostclaim.Claim) error { |
| 157 | if !hostclaim.SameWriterIdentity(admitted, hostclaim.Observed{Host: admitted.Host, Claim: c}) { |
| 158 | return hostbox.ErrChanged |
| 159 | } |
| 160 | return nil |
| 161 | }, func() error { return h.installRecoveryInventory(ctx, prior, admitted) }) |
| 162 | } |
| 163 | |
| 164 | // The first return to the cut installation has not launched under this operation. |
| 165 | // Later source sessions cannot reuse the cut's stop evidence. |
| 166 | func claimInitialSourceRecovery(o hostclaim.Observed, box string) bool { |
| 167 | if !claimRecoveryObservation(o) || o.Claim.Holder == nil { |
| 168 | return false |
| 169 | } |
| 170 | c, h := o.Claim, o.Claim.Holder |
| 171 | cut, e := strconv.ParseUint(c.Op.Journal[hostclaim.JournalCutSession], 10, 64) |
| 172 | stamp, se := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64) |
| 173 | return e == nil && se == nil && stamp > 0 && h.Stamp == stamp && c.Op.ID != "" && c.Op.Phase == hostclaim.PhaseProbing && c.Op.Exec == box && c.Op.Intent["source"] == box && h.Session == cut && (h.WriterBox == box || h.Session == 0 && h.WriterBox == "" && len(h.Admissions) == 0) && h.Generation == c.Op.Journal[hostclaim.JournalSrcGen] |
| 174 | } |
| 175 | func (h claimCrossBoxHandover) sourceRecoveryEvidence(ctx context.Context, o hostclaim.Observed) error { |
| 176 | if !claimInitialSourceRecovery(o, h.Box) || h.Evidence == nil { |
| 177 | return hostbox.ErrChanged |
| 178 | } |
| 179 | c := o.Claim |
| 180 | key := moves.StoppedKey(moves.ReportPrefix(c.Holder.Account, c.Holder.App, o.Host, c.Op.ID), "source") |
| 181 | if c.Op.Journal[hostclaim.JournalEvidenceSource] != key { |
| 182 | return hostbox.ErrChanged |
| 183 | } |
| 184 | report, err := readStoppedReport(ctx, h.Evidence, key) |
| 185 | if err != nil { |
| 186 | return err |
| 187 | } |
| 188 | if report.Role != "source" || report.Box != h.Box || report.Host != o.Host || report.OpID != c.Op.ID || report.Incarnation != c.Holder.Incarnation { |
| 189 | return hostbox.ErrChanged |
| 190 | } |
| 191 | return nil |
| 192 | } |
| 193 | |
| 194 | // The guard is already held. Preserve exact inventory entries and their physical |
| 195 | // identities; only ownership moves to the admitted installation. The old inventory |
| 196 | // remains available for retirement and replay. Never infer an empty installation. |
| 197 | func (h claimCrossBoxHandover) installRecoveryInventory(ctx context.Context, prior, admitted hostclaim.Observed) error { |
| 198 | if h.Inventory == nil { |
| 199 | return hostbox.ErrTypedInventory |
| 200 | } |
| 201 | actor, err := h.Identity.Verify(ctx) |
| 202 | if err != nil { |
| 203 | return err |
| 204 | } |
| 205 | if actor.Box != h.Box { |
| 206 | return hostbox.ErrChanged |
| 207 | } |
| 208 | old, next := prior.Claim.Holder, admitted.Claim.Holder |
| 209 | stamp, err := strconv.ParseUint(prior.Claim.Op.Journal[hostclaim.JournalSourceStamp], 10, 64) |
| 210 | if err != nil || stamp == 0 || stamp >= next.Stamp { |
| 211 | return hostbox.ErrChanged |
| 212 | } |
| 213 | inv, err := h.Inventory.ReadInstallation(ctx, prior.Host, old.Incarnation, stamp) |
| 214 | if err != nil { |
| 215 | return err |
| 216 | } |
| 217 | // The predecessor is reused only under today's binding (roster binding |
| 218 | // R3); one stamped under another binding is refused before any reuse. |
| 219 | if !inv.Complete || inv.Host != prior.Host || inv.Incarnation != old.Incarnation || inv.InstallationStamp != stamp || inv.Box != actor.Box || inv.Enrollment != actor.Enrollment || actor.Binding == "" || inv.DirectoryRevision != actor.Binding { |
| 220 | return hostbox.ErrTypedInventory |
| 221 | } |
| 222 | sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", prior.Host, next.Incarnation, next.Stamp, actor.Box))) |
| 223 | owner := hex.EncodeToString(sum[:]) |
| 224 | current, err := h.Inventory.ReadInstallation(ctx, prior.Host, next.Incarnation, next.Stamp) |
| 225 | if err == nil { |
| 226 | if !current.Complete || current.Box != actor.Box || current.Enrollment != actor.Enrollment || current.DirectoryRevision != actor.Binding { |
| 227 | return hostbox.ErrTypedInventory |
| 228 | } |
| 229 | if h.Registry == nil { |
| 230 | return hostbox.ErrTypedInventory |
| 231 | } |
| 232 | row, e := h.Registry.Get(prior.Host) |
| 233 | if e != nil { |
| 234 | return e |
| 235 | } |
| 236 | if err := hostbox.ValidateAgentInventory(current, row, h.Guard.DataDir); err != nil { |
| 237 | return err |
| 238 | } |
| 239 | inv.InstallationStamp = next.Stamp |
| 240 | return h.rebindRecoveryRows(ctx, &inv, stamp, owner) |
| 241 | } |
| 242 | if !errors.Is(err, sql.ErrNoRows) { |
| 243 | return err |
| 244 | } |
| 245 | inv.InstallationStamp, inv.DirectoryRevision = next.Stamp, actor.Binding |
| 246 | if err := h.rebindRecoveryRows(ctx, &inv, stamp, owner); err != nil { |
| 247 | return err |
| 248 | } |
| 249 | for i := range inv.Entries { |
| 250 | inv.Entries[i].Ownership = owner |
| 251 | } |
| 252 | if h.afterRecoveryRows != nil { |
| 253 | if err := h.afterRecoveryRows(); err != nil { |
| 254 | return err |
| 255 | } |
| 256 | } |
| 257 | return h.Inventory.WriteStopInventory(ctx, inv) |
| 258 | } |
| 259 | |
| 260 | func (h claimCrossBoxHandover) rebindRecoveryRows(ctx context.Context, inv *hostbox.StopInventory, oldStamp uint64, owner string) error { |
| 261 | if h.Registry == nil { |
| 262 | return hostbox.ErrTypedInventory |
| 263 | } |
| 264 | row, err := h.Registry.Get(inv.Host) |
| 265 | if err != nil { |
| 266 | return err |
| 267 | } |
| 268 | if row == nil || row.ClaimIncarnation != inv.Incarnation || (row.ClaimStamp != oldStamp && row.ClaimStamp != inv.InstallationStamp) || !row.ClaimOwnership.Valid() { |
| 269 | return hostbox.ErrTypedInventory |
| 270 | } |
| 271 | timers, err := h.Registry.TimersFor(inv.Host) |
| 272 | if err != nil { |
| 273 | return err |
| 274 | } |
| 275 | old := registry.ColdRows{Route: *row, Timers: append([]registry.Timer(nil), timers...)} |
| 276 | next := registry.ColdRows{Route: *row, Timers: append([]registry.Timer(nil), timers...)} |
| 277 | ownership := func(kind, name string) registry.ClaimOwnership { |
| 278 | sum := sha256.Sum256([]byte(owner + "\x00" + kind + "\x00" + name)) |
| 279 | return registry.ClaimOwnership{ClaimIncarnation: inv.Incarnation, ClaimStamp: inv.InstallationStamp, ClaimOwner: owner, ClaimRowID: hex.EncodeToString(sum[:])} |
| 280 | } |
| 281 | next.Route.ClaimOwnership = ownership("route", inv.Host) |
| 282 | for i := range next.Timers { |
| 283 | next.Timers[i].ClaimOwnership = ownership("timer", next.Timers[i].Name) |
| 284 | } |
| 285 | // Retained source inventory recovers the old row identities after a lost reply. |
| 286 | routeFound := false |
| 287 | for i := range inv.Entries { |
| 288 | e := &inv.Entries[i] |
| 289 | if e.Kind == "route" { |
| 290 | if routeFound || e.RelativePath != inv.Host { |
| 291 | return hostbox.ErrTypedInventory |
| 292 | } |
| 293 | routeFound = true |
| 294 | old.Route.ClaimOwnership = registry.ClaimOwnership{ClaimIncarnation: inv.Incarnation, ClaimStamp: oldStamp, ClaimOwner: e.Ownership, ClaimRowID: e.RowIdentity} |
| 295 | e.RowIdentity = next.Route.ClaimRowID |
| 296 | } |
| 297 | if e.Kind == "timer" { |
| 298 | found := false |
| 299 | for j := range old.Timers { |
| 300 | if e.RelativePath == inv.Host+"/"+old.Timers[j].Name { |
| 301 | old.Timers[j].ClaimOwnership = registry.ClaimOwnership{ClaimIncarnation: inv.Incarnation, ClaimStamp: oldStamp, ClaimOwner: e.Ownership, ClaimRowID: e.RowIdentity} |
| 302 | e.RowIdentity = next.Timers[j].ClaimRowID |
| 303 | found = true |
| 304 | } |
| 305 | } |
| 306 | if !found { |
| 307 | return hostbox.ErrTypedInventory |
| 308 | } |
| 309 | } |
| 310 | } |
| 311 | if !routeFound { |
| 312 | return hostbox.ErrTypedInventory |
| 313 | } |
| 314 | source := *inv |
| 315 | source.InstallationStamp = oldStamp |
| 316 | source.Entries = append([]hostbox.StopInventoryEntry(nil), inv.Entries...) |
| 317 | for i := range source.Entries { |
| 318 | e := &source.Entries[i] |
| 319 | if e.Kind == "route" { |
| 320 | e.RowIdentity = old.Route.ClaimRowID |
| 321 | } |
| 322 | if e.Kind == "timer" { |
| 323 | for _, t := range old.Timers { |
| 324 | if e.RelativePath == inv.Host+"/"+t.Name { |
| 325 | e.RowIdentity = t.ClaimRowID |
| 326 | } |
| 327 | } |
| 328 | } |
| 329 | } |
| 330 | if err := hostbox.ValidateAgentInventory(source, &old.Route, h.Guard.DataDir); err != nil { |
| 331 | return err |
| 332 | } |
| 333 | return h.Registry.RebindClaimRows(ctx, old, next) |
| 334 | } |
| 335 | |
amadan