| new file mode 100644 |
| index 000000000..e6430899d |
| --- /dev/null |
| +++ b/internal/host/movesoffline_return.go |
| @@ -0,0 +1,257 @@ |
| +package host |
| + |
| +// The operator's return, the box's half (self-serve moves plan 2 Task 8; |
| +// spec §7.5). A box that was a source of a completed forward move answers |
| +// each return request with an immutable return-ready report of what the |
| +// forward left on it: the console's stage or start door accepts exactly |
| +// that set (internal/console offlinereturn.go). |
| +// |
| +// The scan: every offlineReturnScanEvery the box lists the forward moves' |
| +// prefixes (control/offline-moves/<host>/<id>/) and reads each open record |
| +// once (cached: an open record is immutable), keeping those it may answer |
| +// for (moves.ReturnParticipant: a forward source participant by box id, |
| +// and by instance id when one was pinned; without participants a box of |
| +// the forward's source region). Every pass it lists the requests of those |
| +// it keeps, takes the current one (moves.CurrentReturnRequest) and, when no |
| +// report of this box answers it yet, writes one create-only. A create-only |
| +// conflict is the request answered already; a newer request gets its own |
| +// report. Nothing is overwritten, so no ordering between writers matters. |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "os" |
| + "path/filepath" |
| + "sort" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/layout" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/regionname" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// offlineReturnScanEvery is how often the box lists the forward moves' |
| +// prefixes for open records it may answer returns of (moveCfg. |
| +// ReturnScanEvery overrides it; negative is every pass). |
| +const offlineReturnScanEvery = time.Minute |
| + |
| +// offlineMovesRoot is the prefix every offline move's reports live under. |
| +const offlineMovesRoot = "control/offline-moves/" |
| + |
| +// returnState is the worker's memory of the scan, under pass. |
| +type returnState struct { |
| + scanned time.Time |
| + // opens is every open record read, by "<host>/<id>"; nil value: no |
| + // open record there (read again on the next scan). |
| + opens map[string]*moves.OpenRecord |
| +} |
| + |
| +// returnScanDue reports whether the forward prefixes are listed this pass. |
| +func (w *movesWorker) returnScanDue(now time.Time) bool { |
| + every := w.cfg.ReturnScanEvery |
| + if every == 0 { |
| + every = offlineReturnScanEvery |
| + } |
| + return every < 0 || w.returns.scanned.IsZero() || now.Sub(w.returns.scanned) >= every |
| +} |
| + |
| +// mayAnswer reports whether this box may answer returns of the forward o. |
| +func (w *movesWorker) mayAnswer(o *moves.OpenRecord) bool { |
| + cfg := w.cfg |
| + if o.Participants != nil { |
| + return moves.ReturnParticipant(*o, cfg.BoxID, cfg.instance(), "") |
| + } |
| + here := cfg.parkRegion() |
| + return here != "" && regionname.PlacementMatches(cfg.ParkHomeRegion, o.From, here) |
| +} |
| + |
| +// offlineReturns is the pass's return step: errors are logged and the next |
| +// pass tries again. |
| +func (w *movesWorker) offlineReturns(ctx context.Context, s release.Store) { |
| + cfg := w.cfg |
| + if cfg.Offline == nil || cfg.BoxID == "" || cfg.OwnBucket == "" { |
| + return |
| + } |
| + now := cfg.now() |
| + if w.returns.opens == nil { |
| + w.returns.opens = map[string]*moves.OpenRecord{} |
| + } |
| + if w.returnScanDue(now) { |
| + if err := w.scanReturns(ctx, s); err != nil { |
| + w.notice("#offline-returns", "moves: the return scan: %v", err) |
| + return |
| + } |
| + w.returns.scanned = now |
| + } |
| + keys := make([]string, 0, len(w.returns.opens)) |
| + for k, o := range w.returns.opens { |
| + if o != nil && w.mayAnswer(o) { |
| + keys = append(keys, k) |
| + } |
| + } |
| + sort.Strings(keys) |
| + for _, k := range keys { |
| + if err := w.answerReturn(ctx, s, w.returns.opens[k]); err != nil { |
| + w.notice(k+"#offline-return", "moves: %s: the return of %s: %v", w.returns.opens[k].Host, w.returns.opens[k].MoveID, err) |
| + } |
| + } |
| +} |
| + |
| +// scanReturns lists the forward prefixes and reads the open records not |
| +// read yet. |
| +func (w *movesWorker) scanReturns(ctx context.Context, s release.Store) error { |
| + hosts, err := s.ListDirs(ctx, offlineMovesRoot) |
| + if err != nil { |
| + return err |
| + } |
| + for _, hp := range hosts { |
| + host := strings.TrimSuffix(strings.TrimPrefix(hp, offlineMovesRoot), "/") |
| + if !provision.ValidHostShape(host) { |
| + continue |
| + } |
| + ids, err := s.ListDirs(ctx, hp) |
| + if err != nil { |
| + return err |
| + } |
| + for _, ip := range ids { |
| + id := strings.TrimSuffix(strings.TrimPrefix(ip, hp), "/") |
| + k := host + "/" + id |
| + if w.returns.opens[k] != nil { |
| + continue |
| + } |
| + key := moves.OpenRecordKey(host, id) |
| + if key == "" { |
| + continue |
| + } |
| + raw, _, err := s.Get(ctx, key) |
| + if errors.Is(err, os.ErrNotExist) { |
| + continue |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + var o moves.OpenRecord |
| + if json.Unmarshal(raw, &o) != nil || o.MoveID != id || o.Host != host || o.OpenedAt.IsZero() { |
| + continue |
| + } |
| + w.returns.opens[k] = &o |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// answerReturn writes this box's report for the current request of the |
| +// forward o, once. |
| +func (w *movesWorker) answerReturn(ctx context.Context, s release.Store, o *moves.OpenRecord) error { |
| + cfg := w.cfg |
| + prefix := moves.ReturnRequestPrefix(o.Host, o.MoveID) |
| + keys, err := s.List(ctx, prefix) |
| + if err != nil { |
| + return err |
| + } |
| + var reqs []moves.ReturnRequest |
| + for _, k := range keys { |
| + raw, _, err := s.Get(ctx, k) |
| + if err != nil { |
| + continue |
| + } |
| + if rq, err := moves.ParseReturnRequest(raw, o.Host, o.MoveID, k); err == nil { |
| + reqs = append(reqs, rq) |
| + } |
| + } |
| + cur, ok := moves.CurrentReturnRequest(reqs) |
| + if !ok { |
| + return nil |
| + } |
| + key := moves.ReturnReportKey(o.Host, o.MoveID, cfg.BoxID, cur.Key) |
| + if key == "" { |
| + return fmt.Errorf("no report key for request %s", cur.Key) |
| + } |
| + if _, _, err := s.Get(ctx, key); err == nil { |
| + return nil // answered |
| + } else if !errors.Is(err, os.ErrNotExist) { |
| + return err |
| + } |
| + rec, err := provision.GetRecord(ctx, s, o.Account, o.App, o.Host) |
| + if err != nil { |
| + return fmt.Errorf("the record: %w", err) |
| + } |
| + // The ready facts as for a move of this host onto this box, the |
| + // forward's id naming the park directory it left. |
| + ret := *rec |
| + ret.Move = &provision.MoveState{Kind: provision.MoveKindOffline, ID: o.MoveID, From: o.To, To: o.From} |
| + ready, err := w.offlineTargetReady(ctx, s, ret) |
| + if err != nil { |
| + return err |
| + } |
| + lo, err := w.returnLeftovers(ctx, o, ready) |
| + if err != nil { |
| + return err |
| + } |
| + ready.Leftovers = &lo |
| + rep := moves.OfflineReport{V: moves.OfflineReportVersion, Op: o.MoveID, Host: o.Host, Box: cfg.Box, Role: moves.OfflineRoleTarget, |
| + Phase: moves.OfflinePhaseReady, At: cfg.now().UTC(), Request: cur.RequestedAt, RequestKey: cur.Key, Ready: &ready, |
| + BoxID: cfg.BoxID, InstanceID: cfg.instance()} |
| + body, err := json.Marshal(rep) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := s.Put(ctx, key, body, ""); err != nil { |
| + if errors.Is(err, release.ErrCASConflict) { |
| + return nil // answered by an earlier pass |
| + } |
| + return err |
| + } |
| + log.Printf("moves: %s: reported the leftovers of move %s for return request %s", o.Host, o.MoveID, cur.Key) |
| + return nil |
| +} |
| + |
| +// returnLeftovers is what the forward o left on this box: the source's |
| +// lock, its park directories, the nested directories left empty, and the |
| +// replica's objects written since the forward opened. |
| +func (w *movesWorker) returnLeftovers(ctx context.Context, o *moves.OpenRecord, ready moves.TargetReadyReport) (moves.ReturnLeftovers, error) { |
| + cfg := w.cfg |
| + lo := moves.ReturnLeftovers{ForwardID: o.MoveID, Layout: o.Layout, ParkDirs: []string{}, EmptyNested: []string{}} |
| + st, err := w.offlineLock(ctx, o.Host) |
| + if err != nil { |
| + return lo, err |
| + } |
| + lo.LockOwner, lo.LockLifted = st.owner, lease.IsLiftedTombstone(st.owner, st.expires) |
| + parkRoot := filepath.Join(cfg.DataDir, ".offline-moves", o.Host) |
| + ids, err := os.ReadDir(parkRoot) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return lo, err |
| + } |
| + for _, id := range ids { |
| + if !id.IsDir() { |
| + lo.ParkDirs = append(lo.ParkDirs, id.Name()) |
| + continue |
| + } |
| + roles, err := os.ReadDir(filepath.Join(parkRoot, id.Name())) |
| + if err != nil { |
| + return lo, err |
| + } |
| + for _, r := range roles { |
| + lo.ParkDirs = append(lo.ParkDirs, id.Name()+"/"+r.Name()+"/") |
| + } |
| + } |
| + for _, p := range []string{provision.InstanceDir(cfg.DataDir, o.Host), layout.RawInstanceDir(cfg.DataDir, o.Host)} { |
| + if entries, err := os.ReadDir(p); err == nil && len(entries) == 0 { |
| + lo.EmptyNested = append(lo.EmptyNested, p) |
| + } |
| + } |
| + for k, obj := range ready.InstancesPrefix.Objects { |
| + if !obj.LastModified.Before(o.OpenedAt) { |
| + lo.NewerObjects = append(lo.NewerObjects, k) |
| + } |
| + } |
| + sort.Strings(lo.NewerObjects) |
| + return lo, nil |
| +} |
amadan