amadan

carlos / platform Public

genselect: the read-only generation selector, Select and SelectPrepared (claim Plan 7b Task 2)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 8c15b9578ae33ea37aa9ff0e936a2ca4a63ee002 parent be741e8
1 file changed, +498 −0
  • internal/genselect/select.go +498 −0
diff --git a/internal/genselect/select.go b/internal/genselect/select.go
new file mode 100644
index 000000000..e956f8c67
--- /dev/null
+++ b/internal/genselect/select.go
@@ -0,0 +1,498 @@
+package genselect
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/superfly/ltx"
+)
+
+var (
+ ErrListing = errors.New("genselect: incomplete or invalid listing")
+ ErrDisagreement = errors.New("genselect: marker and holder disagree")
+ ErrIncarnation = errors.New("genselect: marker incarnation mismatch")
+ ErrMarker = errors.New("genselect: malformed marker")
+ ErrVersionID = errors.New("genselect: immutable version id required")
+ ErrDeleteMarker = errors.New("genselect: selected version is a delete marker")
+ ErrPinnedMissing = errors.New("genselect: pinned snapshot version missing")
+ ErrPinnedMismatch = errors.New("genselect: pinned snapshot mismatch")
+ ErrGap = errors.New("genselect: non-contiguous ltx files")
+ ErrEmpty = errors.New("genselect: no restorable data")
+ ErrRegion = errors.New("genselect: baseline belongs to another region")
+ ErrInput = errors.New("genselect: invalid input")
+ ErrRoot = errors.New("genselect: invalid root")
+)
+
+type Reader interface {
+ ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error)
+ GetVersion(context.Context, string, string, string) ([]byte, error)
+ OpenVersion(context.Context, string, string, string) (io.ReadCloser, error)
+}
+
+var _ Reader = (lease.VersionedStore)(nil)
+
+type Pin struct {
+ SnapshotKey string `json:"snapshot_key"`
+ SnapshotVersion string `json:"snapshot_version"`
+ SnapshotTXID uint64 `json:"snapshot_txid"`
+ SnapshotCRC64 uint64 `json:"snapshot_crc64"`
+}
+
+type Marker struct {
+ Op string `json:"op"`
+ Incarnation string `json:"incarnation"`
+ Host string `json:"host"`
+ Box string `json:"box"`
+ InstalledAt time.Time `json:"installed_at"`
+ ManifestSHA256 string `json:"manifest_sha256"`
+ Pin
+}
+
+// PreparedPin is supplied only from an authorized, completed build journal.
+// The caller binds this exact generation and pin to the authorized holder,
+// operation and destination region before calling SelectPrepared.
+type PreparedPin struct {
+ Generation string
+ SnapshotKey string
+ SnapshotVersion string
+ SnapshotTXID uint64
+ SnapshotCRC64 uint64
+}
+
+type Request struct {
+ Bucket, Region, Host string
+ Holder hostclaim.Holder
+}
+
+type File struct {
+ Key string `json:"key"`
+ VersionID string `json:"version_id"`
+ Level int `json:"level"`
+ MinTXID uint64 `json:"min_txid"`
+ MaxTXID uint64 `json:"max_txid"`
+ Size int64 `json:"size"`
+}
+
+type Plan struct {
+ Generation string
+ Files []File
+ Root *File
+ Marker *Marker
+ MarkerVersion string
+ Pin *Pin
+ MaxTXID uint64
+ MaxLevel0TXID uint64
+}
+
+type candidate struct{ generation, region string }
+
+func origin(g string) string {
+ if g == "" {
+ return hostclaim.RootGeneration
+ }
+ return g
+}
+
+func component(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
+}
+
+func generation(s string) bool { return component(s) && s != hostclaim.RootGeneration }
+
+func versionID(s string) bool { return strings.TrimSpace(s) != "" && s != "null" }
+
+func prefix(location, gen string) string {
+ if gen == hostclaim.RootGeneration {
+ return location
+ }
+ return location + "g-" + gen + "/"
+}
+
+func coordinates(key, base string) (File, bool) {
+ if !strings.HasPrefix(key, base) {
+ return File{}, false
+ }
+ parts := strings.Split(strings.TrimPrefix(key, base), "/")
+ if len(parts) != 2 || len(parts[0]) != 4 {
+ return File{}, false
+ }
+ level, err := strconv.ParseUint(parts[0], 16, 16)
+ if err != nil || level > 9 || fmt.Sprintf("%04x", level) != parts[0] {
+ return File{}, false
+ }
+ min, max, err := ltx.ParseFilename(parts[1])
+ if err != nil || min == 0 || min > max || ltx.FormatFilename(min, max) != parts[1] {
+ return File{}, false
+ }
+ return File{Key: key, Level: int(level), MinTXID: uint64(min), MaxTXID: uint64(max)}, true
+}
+
+func exactFile(v lease.ObjectVersion, base string) (File, error) {
+ f, ok := coordinates(v.Key, base)
+ if !ok || v.Size < 0 {
+ return File{}, ErrListing
+ }
+ if !versionID(v.VersionID) {
+ return File{}, ErrVersionID
+ }
+ if v.DeleteMarker {
+ return File{}, ErrDeleteMarker
+ }
+ f.VersionID, f.Size = v.VersionID, v.Size
+ return f, nil
+}
+
+func requestLocation(s Reader, req Request) (string, error) {
+ h := req.Holder
+ location := strings.TrimSuffix(h.Location, "/")
+ if s == nil || req.Bucket == "" || req.Region == "" || !component(req.Host) || !component(h.Incarnation) ||
+ (location != "instances/"+req.Host && location != "instances/"+req.Host+"/"+h.Incarnation) ||
+ (h.Generation != "" && !generation(h.Generation)) {
+ return "", ErrInput
+ }
+ return location + "/", nil
+}
+
+func listVersions(ctx context.Context, s Reader, bucket, base string) ([]lease.ObjectVersion, map[string]lease.ObjectVersion, error) {
+ // VersionedStore drains all pages before returning. Never use partial data.
+ versions, err := s.ListVersions(ctx, bucket, base)
+ if err != nil {
+ return nil, nil, fmt.Errorf("%w: %w", ErrListing, err)
+ }
+ latest := make(map[string]lease.ObjectVersion)
+ keys := make(map[string]bool)
+ for _, v := range versions {
+ if !strings.HasPrefix(v.Key, base) {
+ return nil, nil, ErrListing
+ }
+ keys[v.Key] = true
+ if v.IsLatest {
+ if _, exists := latest[v.Key]; exists {
+ return nil, nil, ErrListing
+ }
+ latest[v.Key] = v
+ }
+ }
+ for k := range keys {
+ if _, ok := latest[k]; !ok {
+ return nil, nil, ErrListing
+ }
+ }
+ return versions, latest, nil
+}
+
+// Select resolves the ordinary baseline using the marker and unfinished run.
+// It returns identities only. The consumer must verify every complete file.
+func Select(ctx context.Context, s Reader, req Request) (*Plan, error) {
+ location, err := requestLocation(s, req)
+ if err != nil {
+ return nil, err
+ }
+ versions, latest, err := listVersions(ctx, s, req.Bucket, location)
+ if err != nil {
+ return nil, err
+ }
+ h := req.Holder
+ p := &Plan{Files: []File{}}
+ markerGen := hostclaim.RootGeneration
+ if mv, ok := latest[location+"generation.json"]; ok {
+ if !versionID(mv.VersionID) {
+ return nil, ErrVersionID
+ }
+ if mv.DeleteMarker {
+ return nil, ErrDeleteMarker
+ }
+ data, err := s.GetVersion(ctx, req.Bucket, mv.Key, mv.VersionID)
+ if err != nil {
+ return nil, fmt.Errorf("%w: %w", ErrMarker, err)
+ }
+ var fields map[string]json.RawMessage
+ if err := json.Unmarshal(data, &fields); err != nil {
+ return nil, ErrMarker
+ }
+ if raw, ok := fields["snapshot_txid"]; !ok || string(raw) == "null" {
+ return nil, ErrMarker
+ }
+ var m Marker
+ dec := json.NewDecoder(bytes.NewReader(data))
+ // Unknown additive fields are allowed. Malformed/trailing JSON is not.
+ if err := dec.Decode(&m); err != nil {
+ return nil, fmt.Errorf("%w: %w", ErrMarker, err)
+ }
+ if err := dec.Decode(new(any)); err != io.EOF {
+ return nil, ErrMarker
+ }
+ if m.Incarnation != h.Incarnation {
+ return nil, ErrIncarnation
+ }
+ if m.Host != req.Host || !generation(m.Op) || m.Box == "" || m.InstalledAt.IsZero() {
+ return nil, ErrMarker
+ }
+ markerGen, p.Marker, p.MarkerVersion = m.Op, &m, mv.VersionID
+ }
+ run, err := unfinished(h, markerGen, req.Region)
+ if err != nil {
+ return nil, err
+ }
+ selected := candidate{markerGen, req.Region}
+ for _, c := range run {
+ if c.region != "" && c.region != req.Region {
+ return nil, ErrRegion
+ }
+ if c.generation == markerGen || hasObject(latest, location, c.generation) {
+ selected = c
+ break
+ }
+ }
+ p.Generation = selected.generation
+ base := prefix(location, p.Generation)
+ var pin *Pin
+ if p.Marker != nil && p.Generation == markerGen {
+ cp := p.Marker.Pin
+ pin = &cp
+ }
+ return selectFiles(ctx, s, req.Bucket, base, versions, latest, p, pin)
+}
+
+// SelectPrepared restores only the explicitly authorized prepared generation.
+// It never reads the marker or walks admissions, and never chooses a baseline.
+func SelectPrepared(ctx context.Context, s Reader, req Request, prepared PreparedPin) (*Plan, error) {
+ location, err := requestLocation(s, req)
+ if err != nil {
+ return nil, err
+ }
+ if !generation(prepared.Generation) {
+ return nil, ErrInput
+ }
+ // Empty prepared roots require a separate bootstrap protocol, not fallback.
+ if prepared.SnapshotTXID == 0 {
+ return nil, ErrPinnedMismatch
+ }
+ base := prefix(location, prepared.Generation)
+ versions, latest, err := listVersions(ctx, s, req.Bucket, base)
+ if err != nil {
+ return nil, err
+ }
+ pin := &Pin{SnapshotKey: prepared.SnapshotKey, SnapshotVersion: prepared.SnapshotVersion,
+ SnapshotTXID: prepared.SnapshotTXID, SnapshotCRC64: prepared.SnapshotCRC64}
+ p := &Plan{Generation: prepared.Generation, Files: []File{}}
+ return selectFiles(ctx, s, req.Bucket, base, versions, latest, p, pin)
+}
+
+// Both entry points share exact-version root selection and continuation.
+func selectFiles(ctx context.Context, s Reader, bucket, base string, versions []lease.ObjectVersion,
+ latest map[string]lease.ObjectVersion, p *Plan, pin *Pin) (*Plan, error) {
+ if pin != nil {
+ if pin.SnapshotTXID == 0 {
+ if pin.SnapshotKey != "" || pin.SnapshotVersion != "" || pin.SnapshotCRC64 != 0 {
+ return nil, ErrPinnedMismatch
+ }
+ } else {
+ f, ok := coordinates(pin.SnapshotKey, base)
+ if !ok || f.Level != 9 || f.MinTXID != 1 || f.MaxTXID != pin.SnapshotTXID {
+ return nil, ErrPinnedMismatch
+ }
+ if !versionID(pin.SnapshotVersion) {
+ return nil, ErrVersionID
+ }
+ }
+ p.Pin = pin
+ }
+ var current []File
+ for _, v := range latest {
+ f, ok := coordinates(v.Key, base)
+ if !ok && strings.HasPrefix(v.Key, base) {
+ rel := strings.TrimPrefix(v.Key, base)
+ parts := strings.Split(rel, "/")
+ if len(parts) == 2 && len(parts[0]) == 4 && strings.HasSuffix(parts[1], ".ltx") {
+ if level, e := strconv.ParseUint(parts[0], 16, 16); e == nil && level <= 9 {
+ return nil, ErrListing
+ }
+ }
+ }
+ if !ok || v.DeleteMarker {
+ continue
+ } // Retention tombstones are not current files.
+ f, err := exactFile(v, base)
+ if err != nil {
+ return nil, err
+ }
+ if f.Level == 9 && f.MinTXID != 1 {
+ return nil, ErrRoot
+ }
+ current = append(current, f)
+ if f.Level == 0 && f.MaxTXID > p.MaxLevel0TXID {
+ p.MaxLevel0TXID = f.MaxTXID
+ }
+ }
+ floor := uint64(0)
+ if pin != nil {
+ floor = pin.SnapshotTXID
+ }
+ for _, f := range current {
+ // A late duplicate at the pinned key never replaces its acknowledged version.
+ if pin != nil && floor > 0 && f.Key == pin.SnapshotKey {
+ continue
+ }
+ if f.Level == 9 && f.MinTXID == 1 && f.MaxTXID >= floor && (p.Root == nil || better(f, *p.Root)) {
+ cp := f
+ p.Root = &cp
+ }
+ }
+ pinned := false
+ if p.Root == nil && floor > 0 {
+ var pv *lease.ObjectVersion
+ for _, v := range versions {
+ if v.Key == pin.SnapshotKey && v.VersionID == pin.SnapshotVersion {
+ cp := v
+ pv = &cp
+ break
+ }
+ }
+ if pv == nil {
+ return nil, ErrPinnedMissing
+ }
+ f, err := exactFile(*pv, base)
+ if err != nil {
+ return nil, err
+ }
+ p.Root, pinned = &f, true
+ }
+ if p.Root != nil {
+ if err := checkRoot(ctx, s, bucket, *p.Root, pinned); err != nil {
+ return nil, err
+ }
+ p.Files = append(p.Files, *p.Root)
+ p.MaxTXID = p.Root.MaxTXID
+ }
+ for {
+ var best *File
+ for _, f := range current {
+ if f.Level == 9 || p.MaxTXID == ^uint64(0) || !ltx.IsContiguous(ltx.TXID(p.MaxTXID), ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID)) {
+ continue
+ }
+ if best == nil || better(f, *best) {
+ cp := f
+ best = &cp
+ }
+ }
+ if best == nil {
+ break
+ }
+ if p.Root == nil {
+ cp := *best
+ p.Root = &cp
+ }
+ p.Files = append(p.Files, *best)
+ p.MaxTXID = best.MaxTXID
+ }
+ for _, f := range current {
+ if f.Level < 9 && f.MaxTXID > p.MaxTXID {
+ return nil, ErrGap
+ }
+ }
+ if len(p.Files) == 0 {
+ return nil, ErrEmpty
+ }
+ return p, nil
+}
+
+func unfinished(h hostclaim.Holder, marker, region string) ([]candidate, error) {
+ gen := origin(h.Generation)
+ if gen == marker {
+ return []candidate{{marker, region}}, nil
+ }
+ var out []candidate
+ seen := map[string]bool{}
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ a := h.Admissions[i]
+ if origin(a.Generation) != gen || a.Region != region {
+ return nil, ErrDisagreement
+ }
+ if a.Mode == "warm" {
+ if origin(a.Predecessor) != gen || a.PredecessorRegion != region {
+ return nil, ErrDisagreement
+ }
+ continue
+ }
+ if a.Mode != "cold" || a.Covered || !generation(a.Generation) || seen[gen] {
+ return nil, ErrDisagreement
+ }
+ seen[gen] = true
+ out = append(out, candidate{gen, region})
+ gen, region = origin(a.Predecessor), a.PredecessorRegion
+ if gen == marker {
+ return append(out, candidate{gen, region}), nil
+ }
+ if !generation(gen) || region == "" {
+ return nil, ErrDisagreement
+ }
+ }
+ return nil, ErrDisagreement
+}
+
+func hasObject(latest map[string]lease.ObjectVersion, location, gen string) bool {
+ base := prefix(location, gen)
+ for k := range latest {
+ if !strings.HasPrefix(k, base) {
+ continue
+ }
+ if gen == hostclaim.RootGeneration {
+ if _, ok := coordinates(k, base); !ok {
+ continue
+ }
+ }
+ return true
+ }
+ return false
+}
+
+func better(a, b File) bool {
+ if a.MaxTXID != b.MaxTXID {
+ return a.MaxTXID > b.MaxTXID
+ }
+ if a.MinTXID != b.MinTXID {
+ return a.MinTXID < b.MinTXID
+ }
+ return a.Level > b.Level
+}
+
+func checkRoot(ctx context.Context, s Reader, bucket string, f File, pinned bool) error {
+ category := ErrRoot
+ if pinned {
+ category = ErrPinnedMismatch
+ }
+ rc, err := s.OpenVersion(ctx, bucket, f.Key, f.VersionID)
+ if err != nil {
+ if errors.Is(err, lease.ErrIsDeleteMarker) {
+ return fmt.Errorf("%w: %w", ErrDeleteMarker, err)
+ }
+ if pinned && errors.Is(err, lease.ErrMirrorNotFound) {
+ return fmt.Errorf("%w: %w", ErrPinnedMissing, err)
+ }
+ return fmt.Errorf("%w: %w", category, err)
+ }
+ dec := ltx.NewDecoder(rc)
+ err = dec.DecodeHeader()
+ closeErr := rc.Close()
+ if err != nil {
+ return fmt.Errorf("%w: %w", category, err)
+ }
+ if closeErr != nil {
+ return fmt.Errorf("%w: %w", category, closeErr)
+ }
+ h := dec.Header()
+ if uint64(h.MinTXID) != f.MinTXID || uint64(h.MaxTXID) != f.MaxTXID {
+ return category
+ }
+ return nil
+}