c7e0a8f7c · carlos/platform

diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go

index 5bd273fcb..6b700c26b 100644

--- a/internal/console/offlinereturn.go

+++ b/internal/console/offlinereturn.go

@@ -35,6 +35,7 @@ import (

datalayout "github.com/carlosframework/platform/internal/layout"

"github.com/carlosframework/platform/internal/moves"

"github.com/carlosframework/platform/internal/provision"

+ "github.com/carlosframework/platform/internal/release"

)

// offlineReturnKey reports whether key, under a move's report prefix, is

@@ -145,10 +146,46 @@ func (s *Server) handleAPIInstanceOfflineReturn(w http.ResponseWriter, r *http.R

jsonError(w, http.StatusBadGateway, "could not write the move. try again")

return

}

+ if err := s.offlineReturnIndex(ctx, code, d.host, req.ReturnOf, rq.Key); err != nil {

+ log.Printf("console: offline move %s: the return index of %s: %v", d.host, req.ReturnOf, err)

+ jsonError(w, http.StatusBadGateway, "could not write the move. try again")

+ return

+ }

s.bumpSerial(ctx)

writeJSON(w, http.StatusAccepted, offlineReturnView{Message: msgReturnRequested, Key: rq.Key, RequestedAt: now.UnixNano()})

}

+// offlineReturnIndex writes the region-scoped index entry of a pending

+// return (moves.ReturnIndexKey), after its request: create-only, and an

+// entry already there (an earlier request not answered yet) is moved to the

+// newest key if it still is the one read. The box that answers deletes it

+// only if it is still the entry it read, so a request written while it

+// answered an older one is never lost.

+func (s *Server) offlineReturnIndex(ctx context.Context, code, host, returnOf, requestKey string) error {

+ key := moves.ReturnIndexKey(code, host, returnOf)

+ if key == "" {

+ return fmt.Errorf("no index key for %s in %q", returnOf, code)

+ }

+ body, err := json.Marshal(moves.ReturnIndexEntry{Host: host, ReturnOf: returnOf, RequestKey: requestKey})

+ if err != nil {

+ return err

+ }

+ _, err = s.cfg.Releases.Put(ctx, key, body, "")

+ if !errors.Is(err, release.ErrCASConflict) {

+ return err

+ }

+ _, etag, err := s.cfg.Releases.Get(ctx, key)

+ if errors.Is(err, os.ErrNotExist) {

+ _, err = s.cfg.Releases.Put(ctx, key, body, "")

+ return err

+ }

+ if err != nil {

+ return err

+ }

+ _, err = s.cfg.Releases.Put(ctx, key, body, etag)

+ return err

+}

+

// readReturnReport is the return-ready report of the current request of

// returnOf from the target box sel names: it must exist, be written within

// offlineReturnFresh, answer that request (its key and time), come from a

diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go

index 0d6f2239b..10441cab9 100644

--- a/internal/console/offlinereturn_test.go

+++ b/internal/console/offlinereturn_test.go

@@ -120,6 +120,12 @@ func TestOperatorReturnHandshakeAtTheConsole(t *testing.T) {

if len(keys) != 2 {

t.Fatalf("requests %v", keys)

}

+ // The region's index names the forward with the newest request.

+ raw, _, err := w.fx.rel.Get(context.Background(), moves.ReturnIndexKey("uk", offlineHost, returnFwd))

+ var e moves.ReturnIndexEntry

+ if err != nil || json.Unmarshal(raw, &e) != nil || e.RequestKey != k2 || e.ReturnOf != returnFwd || e.Host != offlineHost {

+ t.Fatalf("the index entry: %s %v", raw, err)

+ }

})

t.Run("an open record of another lifetime", func(t *testing.T) {

w := newReturnWorld(t, func(o *moves.OpenRecord) { o.Incarnation++ })

diff --git a/internal/e2e/selfserve/return_test.go b/internal/e2e/selfserve/return_test.go

index 99ba45338..2d76ade94 100644

--- a/internal/e2e/selfserve/return_test.go

+++ b/internal/e2e/selfserve/return_test.go

@@ -97,7 +97,9 @@ func TestOperatorReturnHandshake(t *testing.T) {

if !ok || r1.RequestKey != first || r1.Ready == nil || r1.Ready.Leftovers == nil || r1.Ready.Leftovers.ForwardID != fwd {

t.Fatalf("the first report: %v %+v", ok, r1)

}

- t.Logf("leftovers: %+v", *r1.Ready.Leftovers)

+ if _, _, err := w.raw.Get(w.ctx, moves.ReturnIndexKey("", host.E2EHost, fwd)); err == nil {

+ t.Fatal("the index entry is still there after the report")

+ }

second := w.returnRequest(fwd)

src.ArmOffline() // an agent restart between the request and its report

w.boxes()

diff --git a/internal/host/movesoffline_return.go b/internal/host/movesoffline_return.go

index 6ff586811..ce000b556 100644

--- a/internal/host/movesoffline_return.go

+++ b/internal/host/movesoffline_return.go

@@ -6,16 +6,18 @@ package host

// 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.

+// The scan: once every offlineReturnScanEvery the box lists the return

+// index of its own region (moves.ReturnIndexPrefix: one LIST, whatever the

+// number of hosts), which the console's return door writes beside each

+// request. For each pending return it reads the forward's open record once

+// (cached, and a record it may not answer for cached as a negative), lists

+// that forward's requests, takes the current one

+// (moves.CurrentReturnRequest) and, when no report of this box answers it

+// yet, writes one create-only; then it deletes the index entry if it is

+// still the one it read (a newer request moved it, and the next scan

+// answers that). A create-only conflict is the request answered already;

+// nothing is overwritten, so no ordering between writers matters. Nothing

+// runs between scans.

import (

"context"

@@ -26,7 +28,6 @@ import (

"os"

"path/filepath"

"sort"

- "strings"

"time"

"github.com/carlosframework/platform/internal/layout"

@@ -42,18 +43,17 @@ import (

// 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.

+// returnState is the worker's memory of the scan, under pass: the open

+// records of the pending returns, and the pending returns this box may not

+// answer for, both by "<host>/<forward id>" and both pruned to the index

+// each scan.

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

+ scanned time.Time

+ opens map[string]*moves.OpenRecord

+ negatives map[string]bool

}

-// returnScanDue reports whether the forward prefixes are listed this pass.

+// returnScanDue reports whether the index is listed this pass.

func (w *movesWorker) returnScanDue(now time.Time) bool {

every := w.cfg.ReturnScanEvery

if every == 0 {

@@ -72,84 +72,97 @@ func (w *movesWorker) mayAnswer(o *moves.OpenRecord) bool {

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.

+// offlineReturns is the pass's return step, a scan when one is due:

+// errors are logged and the next scan 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)

- }

+ if !w.returnScanDue(now) {

+ return

}

- 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)

- }

+ w.returns.scanned = now

+ prefix := moves.ReturnIndexPrefix(cfg.Region)

+ if prefix == "" {

+ return

}

-}

-

-// 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)

+ keys, err := s.List(ctx, prefix)

if err != nil {

- return err

+ w.notice("#offline-returns", "moves: the return index: %v", err)

+ return

}

- for _, hp := range hosts {

- host := strings.TrimSuffix(strings.TrimPrefix(hp, offlineMovesRoot), "/")

- if !provision.ValidHostShape(host) {

+ sort.Strings(keys)

+ opens, negatives := map[string]*moves.OpenRecord{}, map[string]bool{}

+ for _, key := range keys {

+ host, fwd, ok := moves.ParseReturnIndexKey(prefix, key)

+ if !ok {

continue

}

- ids, err := s.ListDirs(ctx, hp)

- if err != nil {

- return err

+ k := host + "/" + fwd

+ if w.returns.negatives[k] {

+ negatives[k] = true

+ continue

}

- for _, ip := range ids {

- id := strings.TrimSuffix(strings.TrimPrefix(ip, hp), "/")

- k := host + "/" + id

- if w.returns.opens[k] != nil {

+ o := w.returns.opens[k]

+ if o == nil {

+ if o, err = w.readOpen(ctx, s, host, fwd); err != nil {

+ w.notice(k+"#offline-return", "moves: %s: the open record of %s: %v", host, fwd, err)

continue

}

- key := moves.OpenRecordKey(host, id)

- if key == "" {

+ if o == nil || !w.mayAnswer(o) {

+ negatives[k] = true

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

+ }

+ opens[k] = o

+ if err := w.answerReturn(ctx, s, o, key); err != nil {

+ w.notice(k+"#offline-return", "moves: %s: the return of %s: %v", host, fwd, err)

}

- return nil

+ w.returns.opens, w.returns.negatives = opens, negatives

+}

+

+// readOpen is the forward's open record, nil when absent or not one.

+func (w *movesWorker) readOpen(ctx context.Context, s release.Store, host, id string) (*moves.OpenRecord, error) {

+ raw, _, err := s.Get(ctx, moves.OpenRecordKey(host, id))

+ if errors.Is(err, os.ErrNotExist) {

+ return nil, nil

+ }

+ if err != nil {

+ return nil, err

+ }

+ var o moves.OpenRecord

+ if json.Unmarshal(raw, &o) != nil || o.MoveID != id || o.Host != host || o.OpenedAt.IsZero() {

+ return nil, nil

+ }

+ return &o, 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 {

+// forward o, once, and then deletes the index entry indexKey if it is still

+// the one read before.

+func (w *movesWorker) answerReturn(ctx context.Context, s release.Store, o *moves.OpenRecord, indexKey string) error {

+ _, indexETag, err := s.Get(ctx, indexKey)

+ if errors.Is(err, os.ErrNotExist) {

+ return nil // answered and removed meanwhile

+ }

+ if err != nil {

+ return err

+ }

+ if err := w.writeReturnReport(ctx, s, o); err != nil {

+ return err

+ }

+ if err := release.DeleteIfMatch(ctx, s, indexKey, indexETag); err != nil && !errors.Is(err, release.ErrCASConflict) {

+ return fmt.Errorf("the index entry: %w", err)

+ }

+ return nil

+}

+

+// writeReturnReport writes this box's report for the current request of

+// the forward o when there is none.

+func (w *movesWorker) writeReturnReport(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)

diff --git a/internal/host/movesoffline_return_test.go b/internal/host/movesoffline_return_test.go

new file mode 100644

index 000000000..cec86306a

--- /dev/null

+++ b/internal/host/movesoffline_return_test.go

@@ -0,0 +1,115 @@

+//go:build linux

+

+package host

+

+import (

+ "context"

+ "encoding/json"

+ "fmt"

+ "strings"

+ "sync"

+ "testing"

+ "time"

+

+ "github.com/carlosframework/platform/internal/moves"

+ "github.com/carlosframework/platform/internal/provision"

+ "github.com/carlosframework/platform/internal/release"

+)

+

+// returnCountStore counts the box's bucket requests by kind and prefix.

+type returnCountStore struct {

+ *release.DirStore

+ mu sync.Mutex

+ counts map[string]int

+}

+

+func (c *returnCountStore) count(kind, key string) {

+ c.mu.Lock()

+ defer c.mu.Unlock()

+ if c.counts == nil {

+ c.counts = map[string]int{}

+ }

+ c.counts[kind]++

+ c.counts[kind+" "+key]++

+}

+

+func (c *returnCountStore) List(ctx context.Context, prefix string) ([]string, error) {

+ c.count("List", prefix)

+ return c.DirStore.List(ctx, prefix)

+}

+func (c *returnCountStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {

+ c.count("ListDirs", prefix)

+ return c.DirStore.ListDirs(ctx, prefix)

+}

+func (c *returnCountStore) Get(ctx context.Context, key string) ([]byte, string, error) {

+ c.count("Get", key)

+ return c.DirStore.Get(ctx, key)

+}

+func (c *returnCountStore) reset() {

+ c.mu.Lock()

+ c.counts = map[string]int{}

+ c.mu.Unlock()

+}

+func (c *returnCountStore) n(kind string) int {

+ c.mu.Lock()

+ defer c.mu.Unlock()

+ return c.counts[kind]

+}

+

+// TestReturnScanIsOneListPerMinute: the box's return step costs one LIST of

+// its region's return index per minute, whatever the number of hosts with

+// offline moves, and nothing between scans; a pending return of a forward

+// it was not a source of reads that open record once and is then cached as

+// a negative.

+func TestReturnScanIsOneListPerMinute(t *testing.T) {

+ ctx := context.Background()

+ s := &returnCountStore{DirStore: release.NewDirStore(t.TempDir())}

+ now := time.Date(2026, 10, 23, 9, 0, 0, 0, time.UTC)

+ w := &movesWorker{cfg: moveCfg{Offline: &offlineView{}, Box: "uk-box", BoxID: "uk-box", OwnBucket: "data-uk", Region: "uk",

+ Now: func() time.Time { return now }}}

+ put := func(key string, v any) {

+ body, _ := json.Marshal(v)

+ if _, err := s.DirStore.Put(ctx, key, body, ""); err != nil {

+ t.Fatal(err)

+ }

+ // 50 hosts with offline moves and their reports: none of them pending.

+ for i := 0; i < 50; i++ {

+ host := fmt.Sprintf("h%d.shop.example.com", i)

+ put(moves.OfflineReportPrefix(host, "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa")+"source/ie-box.json", map[string]any{"v": 1})

+ }

+ s.reset()

+ for i := 0; i < 5; i++ {

+ w.offlineReturns(ctx, s)

+ now = now.Add(10 * time.Second)

+ }

+ if s.n("List") != 1 || s.n("ListDirs") != 0 || s.n("Get") != 0 || s.n("List "+moves.ReturnIndexPrefix("uk")) != 1 {

+ t.Fatalf("within a minute: %v", s.counts)

+ }

+ now = now.Add(time.Minute)

+ w.offlineReturns(ctx, s)

+ if s.n("List") != 2 || s.n("Get") != 0 {

+ t.Fatalf("after a minute: %v", s.counts)

+ }

+

+ // A pending return of a forward another box sourced: its open record

+ // once, then a negative.

+ fwd := "mv_bbbbbbbbbbbbbbbbbbbbbbbbbb"

+ host := "h1.shop.example.com"

+ put(moves.OpenRecordKey(host, fwd), moves.OpenRecord{MoveID: fwd, Host: host, Account: "bab", App: "shop", Incarnation: 1, From: "uk", To: "ie",

+ OpenedAt: now.Add(-time.Hour), Participants: &provision.Participants{Source: []provision.Participant{{BoxID: "uk-other", Label: "uk-other"}}}})

+ put(moves.ReturnIndexKey("uk", host, fwd), moves.ReturnIndexEntry{Host: host, ReturnOf: fwd, RequestKey: "1-c-aaaaaaaa"})

+ s.reset()

+ for i := 0; i < 3; i++ {

+ now = now.Add(61 * time.Second)

+ w.offlineReturns(ctx, s)

+ }

+ if s.n("List") != 3 || s.n("Get "+moves.OpenRecordKey(host, fwd)) != 1 || s.n("Get") != 1 {

+ t.Fatalf("a foreign pending return: %v", s.counts)

+ }

+ for k := range s.counts {

+ if strings.HasPrefix(k, "List control/offline-moves/"+host) {

+ t.Fatalf("listed a foreign forward's requests: %v", s.counts)

+ }

+}

diff --git a/internal/moves/offline.go b/internal/moves/offline.go

index 806e10270..58b66cd18 100644

--- a/internal/moves/offline.go

+++ b/internal/moves/offline.go

@@ -9,6 +9,7 @@ import (

"io"

"path/filepath"

"sort"

+ "strings"

"time"

"github.com/carlosframework/platform/internal/provision"

@@ -1166,3 +1167,62 @@ func ReturnParticipant(o OpenRecord, boxID, instanceID, placement string) bool {

}

return false

}

+

+// ReturnIndexEntry is the region-scoped pointer the return door writes

+// beside each request (ReturnIndexKey), so the boxes of the return's

+// target region find pending returns with one listing: the forward and the

+// newest request's key. The box deletes it (if-match) once its report is

+// written; the request and the report stay the record.

+type ReturnIndexEntry struct {

+ Host string `json:"host"`

+ ReturnOf string `json:"return_of"`

+ RequestKey string `json:"request_key"`

+}

+

+// returnIndexRegion is a region code as an index segment: the home

+// placement's empty code is "_home".

+func returnIndexRegion(code string) string {

+ if code == "" {

+ return "_home"

+ }

+ return code

+}

+

+// ReturnIndexPrefix is the index of pending returns into region code:

+// control/offline-moves/@returns/<code>/. "" for an unusable code.

+func ReturnIndexPrefix(code string) string {

+ seg := returnIndexRegion(code)

+ if !offlineSegment(seg) {

+ return ""

+ }

+ return offlineReturnsRoot + seg + "/"

+}

+

+// offlineReturnsRoot is where the return index lives; "@returns" is never a

+// host (provision.ValidHostShape), so it never meets a move's prefix.

+const offlineReturnsRoot = "control/offline-moves/@returns/"

+

+// ReturnIndexKey is one pending return's index entry:

+// control/offline-moves/@returns/<code>/<host>/<forward id>.json.

+func ReturnIndexKey(code, host, forward string) string {

+ p := ReturnIndexPrefix(code)

+ if p == "" || !provision.ValidHostShape(host) || !offlineSegment(forward) {

+ return ""

+ }

+ return p + host + "/" + forward + ".json"

+}

+

+// ParseReturnIndexKey is the host and forward id an index key under prefix

+// names, false for anything else.

+func ParseReturnIndexKey(prefix, key string) (host, forward string, ok bool) {

+ rest, found := strings.CutPrefix(key, prefix)

+ if !found {

+ return "", "", false

+ }

+ host, file, found := strings.Cut(rest, "/")

+ forward, isJSON := strings.CutSuffix(file, ".json")

+ if !found || !isJSON || !provision.ValidHostShape(host) || !offlineSegment(forward) {

+ return "", "", false

+ }

+ return host, forward, true

+}

diff --git a/internal/moves/offline_openrecord_test.go b/internal/moves/offline_openrecord_test.go

index 70af28219..4613139d7 100644

--- a/internal/moves/offline_openrecord_test.go

+++ b/internal/moves/offline_openrecord_test.go

@@ -54,3 +54,32 @@ func TestOpenRecordKeyAndWire(t *testing.T) {

t.Fatalf("nil participants on the wire: %s", raw)

}

+

+// TestReturnKeys: the return's request, report and index keys, and the

+// index never meets a move's prefix ("returns" is no host).

+func TestReturnKeys(t *testing.T) {

+ const host, fwd = "shop.example.com", "mv_abcdefghijklmnopqrstuvwxyz"

+ if provision.ValidHostShape("@returns") {

+ t.Fatal("the index root is a host shape")

+ }

+ if got := ReturnRequestKey(host, fwd, "1-c-aaaaaaaa"); got != "control/offline-moves/shop.example.com/"+fwd+"/return/1-c-aaaaaaaa.json" {

+ t.Errorf("request key %q", got)

+ }

+ if got := ReturnReportKey(host, fwd, "uk-box", "1-c-aaaaaaaa"); got != "control/offline-moves/shop.example.com/"+fwd+"/reports/return-ready-uk-box-1-c-aaaaaaaa.json" {

+ t.Errorf("report key %q", got)

+ }

+ for code, prefix := range map[string]string{"": "control/offline-moves/@returns/_home/", "uk": "control/offline-moves/@returns/uk/"} {

+ if got := ReturnIndexPrefix(code); got != prefix {

+ t.Errorf("index prefix %q: %q", code, got)

+ }

+ k := ReturnIndexKey(code, host, fwd)

+ if h, f, ok := ParseReturnIndexKey(prefix, k); !ok || h != host || f != fwd {

+ t.Errorf("index key %q parses %q %q %v", k, h, f, ok)

+ }

+ at := time.Unix(0, 5)

+ a, b := ReturnRequest{RequestedAt: at, Key: "5-c-aaaaaaaa"}, ReturnRequest{RequestedAt: at, Key: "5-c-bbbbbbbb"}

+ if cur, _ := CurrentReturnRequest([]ReturnRequest{b, a}); cur.Key != b.Key {

+ t.Errorf("current of two at one time: %q", cur.Key)

+ }

+}

carlos / platform Public

moves, host, console: the return index: the door writes one region-scoped entry per pending return, the box lists only its region's index once a minute, caches open records and negatives, and removes the entry once it has answered