From 9f779568125a4dd341e5c622b4217d67a57df1b4 Mon Sep 17 00:00:00 2001 From: Aric Camarata Date: Mon, 26 Oct 2026 15:36:26 -0400 Subject: [PATCH 1/2] feat(ci): restore a node cache from the coordinator over carrier A (P7-CACHE-21) Coordinator pushes the lease's planned restore blobs as digest and chunk frames; the node session routes them to the cache Before hook, which answers offset, re-hashes and publishes a read-through copy. Held copies cost no chunk frames, a corrupt copy or a flipped byte is discarded and sent again, a failed fetch degrades to a cold miss. Refusals are audited. Carrier B (agent pull) is not covered: agentapi cannot yet carry the frames and is outside this ticket's write scope. --- free/ci/cmd/agent/cache_hook.go | 5 +- free/ci/cmd/agent/cache_lifecycle.go | 17 ++ free/ci/internal/agent/session/control.go | 9 +- free/ci/internal/agent/session/inbound.go | 80 ++++++ .../ci/internal/agent/session/inbound_test.go | 100 ++++++++ free/ci/internal/agent/session/lease.go | 13 +- free/ci/internal/cache/remote/coord.go | 3 + .../cache/remote/integration_agent_test.go | 2 +- .../cache/remote/integration_restore_test.go | 240 ++++++++++++++++++ .../cache/remote/integration_ssh_test.go | 2 +- .../internal/cache/remote/integration_test.go | 12 + free/ci/internal/cache/remote/node_restore.go | 162 ++++++++++++ .../ci/internal/cache/remote/restore_coord.go | 146 +++++++++++ .../internal/cache/remote/restore_offsets.go | 126 +++++++++ free/ci/internal/cache/remote/restore_test.go | 223 ++++++++++++++++ free/ci/internal/run/remote_cache_hook.go | 10 + free/ci/internal/transport/restore.go | 49 ++++ free/ci/internal/transport/ssh/restore.go | 60 +++++ free/ci/internal/transport/ssh/run.go | 3 +- free/ci/internal/transport/ssh/serve.go | 5 + 20 files changed, 1259 insertions(+), 8 deletions(-) create mode 100644 free/ci/internal/agent/session/inbound.go create mode 100644 free/ci/internal/agent/session/inbound_test.go create mode 100644 free/ci/internal/cache/remote/integration_restore_test.go create mode 100644 free/ci/internal/cache/remote/node_restore.go create mode 100644 free/ci/internal/cache/remote/restore_coord.go create mode 100644 free/ci/internal/cache/remote/restore_offsets.go create mode 100644 free/ci/internal/cache/remote/restore_test.go create mode 100644 free/ci/internal/transport/restore.go create mode 100644 free/ci/internal/transport/ssh/restore.go diff --git a/free/ci/cmd/agent/cache_hook.go b/free/ci/cmd/agent/cache_hook.go index c9589ae18..dc42ad437 100644 --- a/free/ci/cmd/agent/cache_hook.go +++ b/free/ci/cmd/agent/cache_hook.go @@ -84,6 +84,7 @@ func cacheBefore(ctx context.Context, l protocol.Lease, root session.JobRoot) (s _ = base.Close() } }() + fetchRestores(ctx, copyStore, l) seen := map[string]bool{} for _, restore := range l.Cache.Restore { sp, ok := specs.Get(restore.Spec) @@ -110,8 +111,8 @@ func cacheBefore(ctx context.Context, l protocol.Lease, root session.JobRoot) (s if !valid { break } - // JobHook.Before has no carrier receiver. Until that protocol hook is - // available a missing copy is best effort; never read coordinator paths. + // fetchRestores filled the read-through copies; a blob still missing + // is a cold miss (nil fetch), and coordinator paths are never read. f, err := copyStore.Open(ctx, l.Cache.Namespace, b.SHA256, b.Size, nil) if err != nil { valid = false diff --git a/free/ci/cmd/agent/cache_lifecycle.go b/free/ci/cmd/agent/cache_lifecycle.go index c139c584e..e35d1cea8 100644 --- a/free/ci/cmd/agent/cache_lifecycle.go +++ b/free/ci/cmd/agent/cache_lifecycle.go @@ -196,3 +196,20 @@ func validateCachePlan(plan *protocol.LeaseCache) error { } return nil } + +// fetchRestores takes the lease's restore blobs from the coordinator over the +// session's reverse blob path (carrier A) into the read-through copies. Without a +// feed (carrier B, or no restore) nothing is fetched: copies already held are still +// used, and a missing one is a cold miss that binds nothing. +func fetchRestores(ctx context.Context, copies *remote.ReadThrough, l protocol.Lease) { + feed, ok := session.FeedFrom(ctx) + if !ok { + return + } + defer feed.Close() + var blobs []protocol.CacheBlob + for _, r := range l.Cache.Restore { + blobs = append(blobs, r.Blobs...) + } + (&remote.NodeRestorer{Copies: copies, Feed: feed, Realm: l.Cache.Namespace}).Fetch(ctx, blobs) +} diff --git a/free/ci/internal/agent/session/control.go b/free/ci/internal/agent/session/control.go index 64927e542..83506767d 100644 --- a/free/ci/internal/agent/session/control.go +++ b/free/ci/internal/agent/session/control.go @@ -8,6 +8,7 @@ import ( "time" "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/transport" ) func (s *Session) onMessage(m protocol.Message) (done bool, err error) { @@ -22,9 +23,13 @@ func (s *Session) onMessage(m protocol.Message) (done bool, err error) { return s.onDrain(m, *b) case *protocol.Offset: s.onOffset(m, *b) + case *protocol.Digest: + s.onInbound(m, transport.BlobFrame{Digest: b}) + case *protocol.Chunk: + s.onInbound(m, transport.BlobFrame{Chunk: b}) } - // heartbeat, ack, upgrade, chunk, digest, resume: carried by later - // tickets (P7-NODE-24 upgrade, P7-CACHE-19 cache restore); the session ignores them. + // heartbeat, ack, upgrade, resume: carried by later tickets (P7-NODE-24 upgrade); + // the session ignores them. return false, nil } diff --git a/free/ci/internal/agent/session/inbound.go b/free/ci/internal/agent/session/inbound.go new file mode 100644 index 000000000..355e164c6 --- /dev/null +++ b/free/ci/internal/agent/session/inbound.go @@ -0,0 +1,80 @@ +package session + +// Purpose: route the coordinator's digest and chunk frames to the job hook that +// restores a cache from them (P7-CACHE-21, Epic CACHE D7): the reverse of an +// upload. The hook finds the lease's feed on its context (FeedFrom). +// Inputs: `digest` and `chunk` frames fenced to a running lease and its epoch. +// Outputs: a bounded channel per lease, read through transport.BlobFeed. +// Constraints: only a lease whose plan restores blobs, on a carrier that sends +// coordinator frames (RestoreCarrier), gets a feed; any other frame is dropped. +// The Run goroutine blocks on a full channel only while the hook still reads it +// (backpressure on the carrier, ended by Close, the lease context or shutdown); +// a frame for an unknown lease or another epoch is ignored. + +import ( + "context" + "sync" + + "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/transport" +) + +// inboundBuffer is how many unread frames a lease keeps (32 chunks is under 24 MiB of base64). +const inboundBuffer = 32 + +// RestoreCarrier is implemented by a Carrier that delivers coordinator digest and +// chunk frames to the session (carrier A). A carrier without it gets no feed. +type RestoreCarrier interface{ BlobRestore() bool } + +// BlobRestore says the stream carrier delivers every coordinator frame. +func (*streamCarrier) BlobRestore() bool { return true } + +type feedKey struct{} + +// FeedFrom returns the reverse blob feed of the lease a Before hook runs for. +func FeedFrom(ctx context.Context) (transport.BlobFeed, bool) { + f, ok := ctx.Value(feedKey{}).(*feed) + if !ok { + return nil, false + } + return f, true +} + +type feed struct { + s *Session + al *activeLease + frames chan transport.BlobFrame + closed chan struct{} + once sync.Once +} + +func (s *Session) newFeed(al *activeLease) *feed { + rc, ok := s.cfg.Carrier.(RestoreCarrier) + if !ok || !rc.BlobRestore() || al.lease.Cache == nil || len(al.lease.Cache.Restore) == 0 { + return nil + } + return &feed{s: s, al: al, frames: make(chan transport.BlobFrame, inboundBuffer), closed: make(chan struct{})} +} + +func (f *feed) Frames() <-chan transport.BlobFrame { return f.frames } +func (f *feed) Offset(o protocol.Offset) error { return f.s.send("offset", o, f.al) } +func (f *feed) Close() { f.once.Do(func() { close(f.closed) }) } + +// onInbound hands a digest or chunk frame to the lease's feed. +func (s *Session) onInbound(m protocol.Message, fr transport.BlobFrame) { + l, ok := s.active[m.LeaseID] + if !ok || m.Epoch == nil || *m.Epoch != l.lease.Epoch || l.feed == nil { + return + } + select { + case <-l.feed.closed: + return + default: + } + select { + case l.feed.frames <- fr: + case <-l.feed.closed: + case <-l.ctxDone: + case <-s.closing: + } +} diff --git a/free/ci/internal/agent/session/inbound_test.go b/free/ci/internal/agent/session/inbound_test.go new file mode 100644 index 000000000..035632248 --- /dev/null +++ b/free/ci/internal/agent/session/inbound_test.go @@ -0,0 +1,100 @@ +package session + +import ( + "context" + "testing" + "time" + + "github.com/nself-org/plugins/free/ci/internal/protocol" +) + +// restoreCar is a fake carrier that, like carrier A, delivers coordinator blob frames. +type restoreCar struct{ *fakeCarrier } + +func (c restoreCar) fake() *fakeCarrier { return c.fakeCarrier } +func (restoreCar) BlobRestore() bool { return true } + +func restoreLease(id string, epoch uint64) protocol.Lease { + l := testLease(id, epoch) + l.Cache = &protocol.LeaseCache{Namespace: "ns", Access: "ro", Restore: []protocol.CacheRestore{{Spec: "go", Key: "k", Dir: "go"}}} + return l +} + +// feedHook records what its Before hook sees of the reverse blob path. +type feedHook struct { + got chan []protocol.Digest + noOne chan bool +} + +func newFeedHook() (*feedHook, JobHook) { + f := &feedHook{got: make(chan []protocol.Digest, 1), noOne: make(chan bool, 1)} + return f, JobHook{Name: "feed", Before: func(ctx context.Context, l protocol.Lease, _ JobRoot) (JobSetup, error) { + feed, ok := FeedFrom(ctx) + if !ok { + f.noOne <- true + return JobSetup{}, nil + } + defer feed.Close() + var seen []protocol.Digest + for len(seen) < 2 { + select { + case fr := <-feed.Frames(): + if fr.Digest != nil { + seen = append(seen, *fr.Digest) + _ = feed.Offset(protocol.Offset{Stream: fr.Digest.Stream, SHA256: fr.Digest.SHA256, Offset: 0}) + } + case <-time.After(5 * time.Second): + seen = append(seen, protocol.Digest{SHA256: "timeout"}) + } + } + f.got <- seen + return JobSetup{}, nil + }} +} + +func TestInboundBlobFramesReachTheHookFenced(t *testing.T) { + f, hook := newFeedHook() + car := restoreCar{newFakeCarrier()} + r := startRigWith(t, t.TempDir(), car, okRunner(`{}`), func(c *Config) { c.Hooks = []JobHook{hook} }) + r.handshake() + r.lease(restoreLease("l1", 5)) + dig := func(sha string) protocol.Digest { return protocol.Digest{Stream: "cache", SHA256: sha, Size: 3} } + // none of these may reach the hook: another epoch, another lease + r.car.pushFenced("digest", dig("stale"), 4, "l1") + r.car.pushFenced("digest", dig("other"), 5, "l2") + r.car.pushFenced("digest", dig("one"), 5, "l1") + r.car.pushFenced("chunk", protocol.Chunk{Stream: "cache", Offset: 0, Data: "AAAA", Digest: "one", Final: true}, 5, "l1") + r.car.pushFenced("digest", dig("two"), 5, "l1") + select { + case seen := <-f.got: + if len(seen) != 2 || seen[0].SHA256 != "one" || seen[1].SHA256 != "two" { + t.Fatalf("hook saw %+v", seen) + } + case <-time.After(10 * time.Second): + t.Fatal("hook never saw the frames") + } + off := r.car.next(t, "offset").Body.(protocol.Offset) + if off.SHA256 != "one" { + t.Fatalf("offset answer %+v", off) + } + if m := r.car.next(t, "offset"); m.LeaseID != "l1" || m.Epoch == nil || *m.Epoch != 5 { + t.Fatalf("offset not fenced to the lease: %+v", m) + } +} + +func TestInboundNoFeedWithoutCarrierSupportOrPlan(t *testing.T) { + f, hook := newFeedHook() + r := startRig(t, okRunner(`{}`), func(c *Config) { c.Hooks = []JobHook{hook} }) // a carrier that cannot deliver blobs + r.handshake() + r.lease(restoreLease("l1", 1)) + if !<-f.noOne { + t.Fatal("a feed was offered on a carrier without blob delivery") + } + f2, hook2 := newFeedHook() + r2 := startRigWith(t, t.TempDir(), restoreCar{newFakeCarrier()}, okRunner(`{}`), func(c *Config) { c.Hooks = []JobHook{hook2} }) + r2.handshake() + r2.lease(testLease("l2", 1)) // no cache plan + if !<-f2.noOne { + t.Fatal("a feed was offered to a lease without restore blobs") + } +} diff --git a/free/ci/internal/agent/session/lease.go b/free/ci/internal/agent/session/lease.go index 3588addc9..472ec3197 100644 --- a/free/ci/internal/agent/session/lease.go +++ b/free/ci/internal/agent/session/lease.go @@ -37,6 +37,8 @@ type activeLease struct { lines atomic.Uint64 // safe lw *lineWriter offsets chan protocol.Offset // safe: the coordinator's offset answers (offsets.go) + feed *feed // set before the job goroutine starts: the reverse blob feed (inbound.go) + ctxDone <-chan struct{} // the lease context's Done channel } func (l *activeLease) cancel() { l.cancelled.Store(true); l.cancelCtx() } @@ -98,6 +100,8 @@ func (s *Session) onLease(m protocol.Message, l protocol.Lease) { // cancel, a drain deadline or the agent fence (P7-NODE-09). ctx, cancel := context.WithCancel(s.runCtx) al := &activeLease{lease: l, cancelCtx: cancel, offsets: make(chan protocol.Offset, offsetBuffer), secWake: make(chan struct{}, 1), secVals: map[string]string{}, expires: exp, secSeen: map[string]bool{}} + al.ctxDone = ctx.Done() + al.feed = s.newFeed(al) if l.RunnerCredential != nil { al.addMask(l.RunnerCredential.Value) } @@ -181,11 +185,15 @@ func (s *Session) runLease(ctx context.Context, al *activeLease, slot *slots.Slo var setup JobSetup skip := "" + beforeCtx := ctx + if al.feed != nil { + beforeCtx = context.WithValue(ctx, feedKey{}, al.feed) + } for _, h := range s.jobHooks { if h.Before == nil { continue } - got, err := h.Before(ctx, l, root) + got, err := h.Before(beforeCtx, l, root) if err != nil && ctx.Err() != nil { skip = "cancelled" // the hook stopped because the lease was cancelled break @@ -196,6 +204,9 @@ func (s *Session) runLease(ctx context.Context, al *activeLease, slot *slots.Slo } mergeSetup(&setup, got) } + if al.feed != nil { + al.feed.Close() // the hooks are done: later frames are dropped + } waitCtx, stopWait := context.WithTimeout(ctx, max(al.expires.Sub(s.cfg.Clock.Now()), 0)) for waitCtx.Err() == nil { al.secMu.Lock() diff --git a/free/ci/internal/cache/remote/coord.go b/free/ci/internal/cache/remote/coord.go index 0e15e2fc5..8143a3da8 100644 --- a/free/ci/internal/cache/remote/coord.go +++ b/free/ci/internal/cache/remote/coord.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "sync" + "time" "github.com/nself-org/plugins/free/ci/internal/authz" "github.com/nself-org/plugins/free/ci/internal/blob/cas" @@ -48,6 +49,8 @@ type Binding struct { Fence, Ready func(context.Context) error // Unordered preserves the full ref grace for carriers without ordered delivery. Unordered bool + // RestoreWait bounds each wait for the node's offset answer (default DefaultRestoreWait). + RestoreWait time.Duration } type Lease struct { diff --git a/free/ci/internal/cache/remote/integration_agent_test.go b/free/ci/internal/cache/remote/integration_agent_test.go index 7d7a80f0f..3ed07a4a0 100644 --- a/free/ci/internal/cache/remote/integration_agent_test.go +++ b/free/ci/internal/cache/remote/integration_agent_test.go @@ -113,7 +113,7 @@ func (a *agentSide) addNode(id string) (model.Capability, tls.Certificate) { } // runAgent delivers the rig's lease to a node over the agent pull carrier and returns the driver's result. -func (r *rig) runAgent(n *node) error { +func (r *rig) runAgent(n nodeSide) error { r.t.Helper() side := r.agentSide() capab, cert := side.addNode("n-1") diff --git a/free/ci/internal/cache/remote/integration_restore_test.go b/free/ci/internal/cache/remote/integration_restore_test.go new file mode 100644 index 000000000..08b3f3a7b --- /dev/null +++ b/free/ci/internal/cache/remote/integration_restore_test.go @@ -0,0 +1,240 @@ +//go:build integration + +package remote + +// Purpose: the cold restore of a node from the coordinator over a real carrier +// (P7-CACHE-21): the real ssh driver pushes digest and chunk frames, a real agent +// session routes them to the Before hook, which restores through NodeRestorer. +// Carrier B (agent pull) cannot send coordinator blob frames yet (see the PR): its +// test pins the degrade path (no feed, a cold miss, no stall). + +import ( + "bytes" + "context" + "encoding/base64" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/nself-org/plugins/free/ci/internal/agent/session" + "github.com/nself-org/plugins/free/ci/internal/protocol" +) + +// restoreNode is node B: it keeps its read-through copies between runs. +type restoreNode struct { + t *testing.T + copies *ReadThrough + carrier bool // carrier A: the session offers the reverse blob feed + flip func(*protocol.Chunk) + chunks atomic.Int32 + noFeed atomic.Bool + atStart atomic.Bool // a verified copy was bound when the job started + data []byte +} + +func newRestoreNode(t *testing.T, carrierA bool) *restoreNode { + t.Helper() + c, err := NewReadThrough(t.TempDir(), 64<<20) + if err != nil { + t.Fatal(err) + } + return &restoreNode{t: t, copies: c, carrier: carrierA} +} + +type countingCarrier struct { + session.Carrier + n *restoreNode +} + +func (c countingCarrier) Recv() (protocol.Message, error) { + m, err := c.Carrier.Recv() + if ch, ok := m.Body.(*protocol.Chunk); ok && err == nil && ch.Stream == StreamCache { + c.n.chunks.Add(1) + if c.n.flip != nil { + c.n.flip(ch) + } + } + return m, err +} + +// carrierACounting keeps the stream carrier's restore capability. +type carrierACounting struct{ countingCarrier } + +func (carrierACounting) BlobRestore() bool { return true } + +func (n *restoreNode) tap(c session.Carrier) session.Carrier { + if n.carrier { + return carrierACounting{countingCarrier{c, n}} + } + return countingCarrier{c, n} +} + +func (n *restoreNode) runner() session.Runner { + return session.RunnerFunc(func(ctx context.Context, jr session.JobRun) (session.JobResult, error) { + b := jr.Lease.Cache.Restore[0].Blobs[0] + if f, err := n.copies.Open(ctx, jr.Lease.Cache.Namespace, b.SHA256, b.Size, nil); err == nil { + _ = f.Close() + n.atStart.Store(true) + } + return session.JobResult{Outcome: session.Outcome{State: "passed"}, Evidence: []byte(`{"schema":"test","ok":true}`)}, nil + }) +} + +func (n *restoreNode) hooks() []session.JobHook { + return []session.JobHook{{Name: "cache", Before: func(ctx context.Context, l protocol.Lease, _ session.JobRoot) (session.JobSetup, error) { + feed, ok := session.FeedFrom(ctx) + if !ok { + n.noFeed.Store(true) + return session.JobSetup{}, nil + } + defer feed.Close() + var blobs []protocol.CacheBlob + for _, r := range l.Cache.Restore { + blobs = append(blobs, r.Blobs...) + } + (&NodeRestorer{Copies: n.copies, Feed: feed, Realm: l.Cache.Namespace, Wait: 10 * time.Second}).Fetch(ctx, blobs) + return session.JobSetup{}, nil + }}} +} + +// planRestore stores data on the coordinator and plans it as the lease's restore. +func (r *rig) planRestore(data []byte) protocol.CacheBlob { + r.t.Helper() + sha, size, err := r.blobs.Put(context.Background(), cacheRealm(r.lease.b.Plan.Namespace), bytes.NewReader(data)) + if err != nil { + r.t.Fatal(err) + } + b := protocol.CacheBlob{Path: "mod", SHA256: sha, Size: size} + r.lease.b.Plan.Restore = []protocol.CacheRestore{{Spec: "go", Key: strings.Repeat("a", 64), Dir: "go", Blobs: []protocol.CacheBlob{b}}} + r.lease.b.RestoreWait = 10 * time.Second + return b +} + +func (n *restoreNode) assertCopy(r *rig, b protocol.CacheBlob) { + n.t.Helper() + f, err := n.copies.Open(context.Background(), r.lease.b.Plan.Namespace, b.SHA256, b.Size, nil) + if err != nil { + n.t.Fatalf("node has no verified copy: %v", err) + } + _ = f.Close() +} + +func TestRemoteCacheRestoreSSH(t *testing.T) { + data := carrierCacheArchive(t) + node := newRestoreNode(t, true) + r1 := newRig(t) + b := r1.planRestore(data) + if err := r1.runSSH(node); err != nil { + t.Fatal(err) + } + if node.noFeed.Load() || !node.atStart.Load() { + t.Fatalf("node B did not hold a verified copy when the job started (noFeed=%v)", node.noFeed.Load()) + } + if node.chunks.Load() < 2 { + t.Fatalf("cold restore moved %d chunk frames", node.chunks.Load()) + } + node.assertCopy(r1, b) + rows, err := r1.st.CacheLocality(context.Background(), "n-1", r1.lease.b.Plan.Namespace, strings.Repeat("a", 64)) + if err != nil || len(rows) != 1 { + t.Fatalf("locality=%+v err=%v", rows, err) + } + // The second run on node B, against a fresh coordinator state, uses its own copy. + before := node.chunks.Load() + node.atStart.Store(false) + r2 := newRig(t) + r2.planRestore(data) + if err = r2.runSSH(node); err != nil { + t.Fatal(err) + } + if !node.atStart.Load() || node.chunks.Load() != before { + t.Fatalf("second run fetched again: %d chunk frames since the first", node.chunks.Load()-before) + } +} + +func TestRemoteCacheRestoreCorrupt(t *testing.T) { + data := carrierCacheArchive(t) + node := newRestoreNode(t, true) + r1 := newRig(t) + b := r1.planRestore(data) + if err := r1.runSSH(node); err != nil { + t.Fatal(err) + } + // A copy that rots on the node is discarded and fetched again. + name, err := node.copies.name(r1.lease.b.Plan.Namespace, b.SHA256) + if err != nil { + t.Fatal(err) + } + raw, _ := os.ReadFile(name) + raw[11] ^= 0xff + if err = os.WriteFile(name, raw, 0o600); err != nil { + t.Fatal(err) + } + before := node.chunks.Load() + node.atStart.Store(false) + r2 := newRig(t) + r2.planRestore(data) + if err = r2.runSSH(node); err != nil { + t.Fatal(err) + } + if !node.atStart.Load() || node.chunks.Load() <= before { + t.Fatalf("corrupt copy was not refetched (chunks %d -> %d)", before, node.chunks.Load()) + } + // A byte flipped in transit fails the re-hash and the blob is sent again. + if err = os.Remove(name); err != nil { + t.Fatal(err) + } + node.flip = flipOnce() + before = node.chunks.Load() + node.atStart.Store(false) + r3 := newRig(t) + r3.planRestore(data) + if err = r3.runSSH(node); err != nil { + t.Fatal(err) + } + if !node.atStart.Load() { + t.Fatal("job started without a verified copy after a flipped byte") + } + if got := node.chunks.Load() - before; got < 4 { + t.Fatalf("a flipped byte did not cause a second pass: %d chunk frames", got) + } + node.assertCopy(r3, b) + // Always flipped: the job still runs, cold, with nothing bound. + if err = os.Remove(name); err != nil { + t.Fatal(err) + } + node.flip = func(c *protocol.Chunk) { + if c.Final { + raw, _ := base64.StdEncoding.DecodeString(c.Data) + raw[0] ^= 1 + c.Data = base64.StdEncoding.EncodeToString(raw) + } + } + node.atStart.Store(false) + r4 := newRig(t) + r4.planRestore(data) + if err = r4.runSSH(node); err != nil { + t.Fatal(err) + } + if node.atStart.Load() { + t.Fatal("a corrupt transfer was bound") + } +} + +func TestRemoteCacheRestoreAgent(t *testing.T) { + data := carrierCacheArchive(t) + node := newRestoreNode(t, false) + r := newRig(t) + r.planRestore(data) + start := time.Now() + if err := r.runAgent(node); err != nil { + t.Fatal(err) + } + if !node.noFeed.Load() || node.atStart.Load() || node.chunks.Load() != 0 { + t.Fatalf("carrier B must degrade to a cold miss without a feed: noFeed=%v bound=%v chunks=%d", node.noFeed.Load(), node.atStart.Load(), node.chunks.Load()) + } + if time.Since(start) > 20*time.Second { + t.Fatal("the cold miss stalled the job") + } +} diff --git a/free/ci/internal/cache/remote/integration_ssh_test.go b/free/ci/internal/cache/remote/integration_ssh_test.go index 20c39543c..3fed8440d 100644 --- a/free/ci/internal/cache/remote/integration_ssh_test.go +++ b/free/ci/internal/cache/remote/integration_ssh_test.go @@ -44,7 +44,7 @@ func sshNode(t *testing.T) (model.Capability, string) { } // runSSH delivers the rig's lease to a node over the ssh driver and returns the driver's result. -func (r *rig) runSSH(n *node) error { +func (r *rig) runSSH(n nodeSide) error { r.t.Helper() capab, home := sshNode(r.t) state := r.t.TempDir() diff --git a/free/ci/internal/cache/remote/integration_test.go b/free/ci/internal/cache/remote/integration_test.go index 40de01dc1..4f3f72196 100644 --- a/free/ci/internal/cache/remote/integration_test.go +++ b/free/ci/internal/cache/remote/integration_test.go @@ -124,6 +124,18 @@ func (s *simSink) Blob(ctx context.Context, h transport.BlobHeader) (transport.B return s.r.lease.Blob(ctx, h) } +// nodeSide is what a carrier needs from the simulated node. +type nodeSide interface { + tap(session.Carrier) session.Carrier + runner() session.Runner + hooks() []session.JobHook +} + +// Restore lets the coordinator push the lease's restore blobs (transport.Restorer). +func (s *simSink) Restore(ctx context.Context, link transport.RestoreLink) error { + return s.r.lease.Restore(ctx, link) +} + type node struct { t *testing.T data []byte diff --git a/free/ci/internal/cache/remote/node_restore.go b/free/ci/internal/cache/remote/node_restore.go new file mode 100644 index 000000000..719048542 --- /dev/null +++ b/free/ci/internal/cache/remote/node_restore.go @@ -0,0 +1,162 @@ +package remote + +// Purpose: the node half of a cache restore (P7-CACHE-21, Epic CACHE D7): answer +// the coordinator's `digest` announces from the read-through copies, take the +// `chunk` frames of a blob the node lacks, and publish a blob only after its +// re-hash matches. +// Inputs: the lease's restore blobs, a transport.BlobFeed and a ReadThrough. +// Outputs: verified copies in the read-through store; `offset` answers (the size +// for a copy held, 0 before a fetch, 0 again after a failed re-hash). +// Constraints: only a blob of the lease plan, with the planned size, is taken: any +// other announce or chunk is ignored; a held copy is re-hashed on every use (a bad +// one is discarded and fetched again); a fetch failure leaves no copy, so the +// caller binds nothing and the job runs cold; every wait is bounded by the +// context and by Wait; a blob is retried at most maxNodeTries times. + +import ( + "context" + "encoding/base64" + "errors" + "io" + "strings" + "time" + + "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/transport" +) + +// DefaultNodeRestoreWait bounds the wait for each coordinator frame. +const DefaultNodeRestoreWait = 20 * time.Second + +const maxNodeTries = 4 + +var errRestart = errors.New("E714 interrupted: the coordinator restarted the blob") + +// NodeRestorer receives a lease's restore blobs from the coordinator. +type NodeRestorer struct { + Copies *ReadThrough + Feed transport.BlobFeed + // Realm is the lease's cache namespace (the read-through realm). + Realm string + // Wait overrides DefaultNodeRestoreWait. + Wait time.Duration + + carry *transport.BlobFrame // a re-announce that cut a stream: handled next +} + +// Fetch returns the digests of the blobs now held as verified copies. It returns +// when every blob is held, a blob ran out of tries, or a frame did not arrive. +func (n *NodeRestorer) Fetch(ctx context.Context, blobs []protocol.CacheBlob) map[string]bool { + got := map[string]bool{} + pending := map[string]protocol.CacheBlob{} + for _, b := range blobs { + pending[b.SHA256] = b + } + tries := map[string]int{} + for len(pending) > 0 { + fr, err := n.next(ctx) + if err != nil { + return got + } + if fr.Digest == nil { + continue // a chunk with no open blob + } + sha := strings.ToLower(strings.TrimPrefix(fr.Digest.SHA256, "sha256:")) + b, ok := pending[sha] + if !ok || fr.Digest.Stream != StreamCache || fr.Digest.Size != b.Size { + continue // outside the plan: never answered, never stored + } + if tries[sha]++; tries[sha] > maxNodeTries { + delete(pending, sha) + continue + } + if n.receive(ctx, b) { + got[sha] = true + delete(pending, sha) + } + } + return got +} + +func (n *NodeRestorer) wait() time.Duration { + if n.Wait > 0 { + return n.Wait + } + return DefaultNodeRestoreWait +} + +func (n *NodeRestorer) next(ctx context.Context) (transport.BlobFrame, error) { + if n.carry != nil { + fr := *n.carry + n.carry = nil + return fr, nil + } + timer := time.NewTimer(n.wait()) + defer timer.Stop() + select { + case fr := <-n.Feed.Frames(): + return fr, nil + case <-ctx.Done(): + return transport.BlobFrame{}, ctx.Err() + case <-timer.C: + return transport.BlobFrame{}, errors.New("E714 interrupted: no frame from the coordinator") + } +} + +func (n *NodeRestorer) answer(b protocol.CacheBlob, off int64) error { + return n.Feed.Offset(protocol.Offset{Stream: StreamCache, SHA256: b.SHA256, Offset: off}) +} + +// receive makes one announced blob a verified copy; false leaves nothing behind. +func (n *NodeRestorer) receive(ctx context.Context, b protocol.CacheBlob) bool { + f, err := n.Copies.Open(ctx, n.Realm, b.SHA256, b.Size, func(ctx context.Context, w io.Writer) error { + if err := n.answer(b, 0); err != nil { // nothing held (or a copy that failed its re-hash): send it all + return err + } + return n.stream(ctx, w, b) + }) + if err != nil { + if ctx.Err() == nil { + _ = n.answer(b, 0) // tells a sender that already finished to send the blob again + } + return false + } + _ = f.Close() + return n.answer(b, b.Size) == nil +} + +// stream writes the chunks of b to w until the final one. +func (n *NodeRestorer) stream(ctx context.Context, w io.Writer, b protocol.CacheBlob) error { + var next int64 + for { + fr, err := n.next(ctx) + if err != nil { + return err + } + if fr.Digest != nil { + if strings.TrimPrefix(fr.Digest.SHA256, "sha256:") == b.SHA256 { + n.carry = &fr + return errRestart + } + continue + } + c := fr.Chunk + if c == nil || c.Stream != StreamCache || strings.TrimPrefix(c.Digest, "sha256:") != b.SHA256 { + continue + } + if c.Offset != next { + return errRestart + } + data, err := base64.StdEncoding.Strict().DecodeString(c.Data) + if err != nil { + return errRestart + } + if _, err = w.Write(data); err != nil { + return err + } + next += int64(len(data)) + if c.Final { + return nil + } + } +} diff --git a/free/ci/internal/cache/remote/restore_coord.go b/free/ci/internal/cache/remote/restore_coord.go new file mode 100644 index 000000000..7b0daa9fb --- /dev/null +++ b/free/ci/internal/cache/remote/restore_coord.go @@ -0,0 +1,146 @@ +package remote + +// Purpose: the coordinator half of a cache restore (P7-CACHE-21, Epic CACHE D7): +// push the blobs the lease plan restores to the node over the reverse blob path +// (digest, then chunk frames; the node answers offset), and report the copies +// the node confirmed to cache_locality. +// Inputs: the lease's Binding (plan, epoch fence, node) and a transport.RestoreLink. +// Outputs: digest and chunk frames; locality rows; audit rows for refusals. +// Constraints: only a blob named by the lease plan, in the plan realm, and only +// while the lease epoch is current and the attempt is leased or running; a blob +// is done when the node confirms it holds every byte (a held copy answers at the +// announce, a fetched one after its re-hash). A node that reports a failed +// re-hash after the final chunk gets the blob again, as many times as the +// transfer allows. Locality is advisory and never authorises a read. + +import ( + "context" + "errors" + "fmt" + "io" + "time" + + "github.com/nself-org/plugins/free/ci/internal/blob/transfer" + "github.com/nself-org/plugins/free/ci/internal/cache/entries" + "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/store" + "github.com/nself-org/plugins/free/ci/internal/transport" +) + +// DefaultRestoreWait bounds each wait for the node's answer. +const DefaultRestoreWait = 30 * time.Second + +// restorePlanned returns the plan entry of a restore blob. +func (l *Lease) restorePlanned(sha string) (protocol.CacheBlob, bool) { + if l.b.Plan == nil { + return protocol.CacheBlob{}, false + } + for _, r := range l.b.Plan.Restore { + for _, b := range r.Blobs { + if b.SHA256 == sha { + return b, true + } + } + } + return protocol.CacheBlob{}, false +} + +// OpenRestore opens one planned restore blob for sending. Every refusal leaves +// an audit row: a digest outside the plan, a stale epoch, a lease no longer +// running, or a cancelled request. +func (l *Lease) OpenRestore(ctx context.Context, sha string) (io.ReadCloser, protocol.CacheBlob, error) { + b, ok := l.restorePlanned(sha) + if !ok { + return nil, b, l.refuse(ctx, entries.ReasonNotPlanned) + } + if ctx.Err() != nil { + return nil, b, l.refuse(context.WithoutCancel(ctx), "cancelled") + } + if err := l.live(ctx); err != nil { + if errors.Is(err, transfer.ErrStale) { + return nil, b, l.refuse(ctx, transfer.ReasonStaleEpoch) + } + return nil, b, err + } + state, _, err := l.h.Ledger.AttemptStateEpoch(ctx, l.b.AttemptID) + if err != nil { + return nil, b, err + } + if state != "leased" && state != "running" { + return nil, b, l.refuse(ctx, transfer.ReasonStaleEpoch) + } + realm := cacheRealm(l.b.Plan.Namespace) + info, err := l.h.Blobs.Stat(realm, sha) + if err != nil { + return nil, b, err + } + if info.Size() != b.Size { + return nil, b, transfer.Failf(transfer.ReasonDigest, "the coordinator holds blob %s at %d B, the plan says %d B", sha[:12], info.Size(), b.Size) + } + rc, err := l.h.Blobs.Open(realm, sha) // re-hashes into a snapshot before the first byte is served + return rc, b, err +} + +// Restore sends the plan's restore blobs to the node, one entry at a time, and +// records the entries the node confirmed. It returns the first error; the node +// degrades the entries that did not arrive to a cold miss. +func (l *Lease) Restore(ctx context.Context, link transport.RestoreLink) error { + if l.b.Plan == nil || len(l.b.Plan.Restore) == 0 { + return nil + } + wait := l.b.RestoreWait + if wait <= 0 { + wait = DefaultRestoreWait + } + tee := newOffsetTee(ctx, link.Offsets()) + up := &finalSnap{link: link, tee: tee} + snd := &transfer.Sender{Up: up, Offsets: tee.out, WaitOffset: wait} + done := map[string]bool{} + for _, r := range l.b.Plan.Restore { + var total int64 + for _, b := range r.Blobs { + total += b.Size + if done[b.SHA256] { + continue + } + if err := l.sendRestore(ctx, snd, up, tee, b, wait); err != nil { + return err + } + done[b.SHA256] = true + } + if l.b.Node != "" { + row := store.CacheLocalityRow{Node: l.b.Node, Realm: l.b.Plan.Namespace, Key: r.Key, Bytes: total} + if err := l.h.Ledger.CacheLocalityPut(ctx, row); err != nil && l.h.Warn != nil { + l.h.Warn(fmt.Errorf("cache locality: %w", err)) + } + } + } + return nil +} + +func (l *Lease) sendRestore(ctx context.Context, snd *transfer.Sender, up *finalSnap, tee *offsetTee, b protocol.CacheBlob, wait time.Duration) error { + blob := transfer.Blob{Stream: StreamCache, SHA256: b.SHA256, Size: b.Size, Open: func() (io.ReadCloser, error) { + rc, _, err := l.OpenRestore(ctx, b.SHA256) + return rc, err + }} + tries := transfer.MaxResumes + 1 + for i := 0; i < tries; i++ { + up.reset() + if err := snd.Send(ctx, blob); err != nil { + return err + } + since, streamed := up.final() + if !streamed { // the node held a verified copy: its announce answer was the whole offset + return nil + } + got, err := tee.after(ctx, b.SHA256, since, wait) + if err != nil { + return err + } + if got == b.Size { + return nil + } + // The node re-hashed the bytes, found them wrong, discarded them and reset to 0: send again. + } + return transfer.Failf(transfer.ReasonDigest, "blob %s failed verification on the node %d times", b.SHA256[:12], tries) +} diff --git a/free/ci/internal/cache/remote/restore_offsets.go b/free/ci/internal/cache/remote/restore_offsets.go new file mode 100644 index 000000000..7c34e79db --- /dev/null +++ b/free/ci/internal/cache/remote/restore_offsets.go @@ -0,0 +1,126 @@ +package remote + +// Purpose: watch the node's offset answers while the coordinator pushes a restore +// blob, so the sender can tell "the node holds every byte" from "the node threw +// the bytes away" after the final chunk (P7-CACHE-21). +// Inputs: the lease's offset channel. +// Outputs: a copy of every answer for the transfer.Sender, and the newest +// position per blob with a count that orders the answers. +// Constraints: forwarding never blocks (a full channel drops the oldest answer); +// the tee stops with its context or when the input closes. + +import ( + "context" + "sync" + "time" + + "github.com/nself-org/plugins/free/ci/internal/blob/transfer" + "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/transport" +) + +type seenOffset struct { + last int64 + n int +} + +type offsetTee struct { + out chan protocol.Offset + mu sync.Mutex + seen map[string]seenOffset + changed chan struct{} +} + +func newOffsetTee(ctx context.Context, in <-chan protocol.Offset) *offsetTee { + t := &offsetTee{out: make(chan protocol.Offset, 64), seen: map[string]seenOffset{}, changed: make(chan struct{})} + go func() { + for { + select { + case <-ctx.Done(): + return + case o, ok := <-in: + if !ok { + return + } + t.note(o) + t.forward(o) + } + } + }() + return t +} + +func (t *offsetTee) note(o protocol.Offset) { + t.mu.Lock() + s := t.seen[o.SHA256] + t.seen[o.SHA256] = seenOffset{last: o.Offset, n: s.n + 1} + close(t.changed) + t.changed = make(chan struct{}) + t.mu.Unlock() +} + +func (t *offsetTee) forward(o protocol.Offset) { + for { + select { + case t.out <- o: + return + default: + } + select { + case <-t.out: + default: + } + } +} + +func (t *offsetTee) count(sha string) int { + t.mu.Lock() + defer t.mu.Unlock() + return t.seen[sha].n +} + +// after waits for an answer about sha newer than count since and returns its offset. +func (t *offsetTee) after(ctx context.Context, sha string, since int, wait time.Duration) (int64, error) { + timer := time.NewTimer(wait) + defer timer.Stop() + for { + t.mu.Lock() + s, ch := t.seen[sha], t.changed + t.mu.Unlock() + if s.n > since { + return s.last, nil + } + select { + case <-ch: + case <-ctx.Done(): + return 0, transfer.Failf(transfer.ReasonInterrupted, "cancelled while waiting for the node to confirm blob %s", sha[:12]) + case <-timer.C: + return 0, transfer.Failf(transfer.ReasonInterrupted, "the node did not confirm blob %s within %s", sha[:12], wait) + } + } +} + +// finalSnap wraps the RestoreLink as a transfer.Uploader and remembers how many +// answers the node had given when the final chunk left. +type finalSnap struct { + link transport.RestoreLink + tee *offsetTee + mu sync.Mutex + since int + sent bool +} + +func (f *finalSnap) reset() { f.mu.Lock(); f.sent = false; f.mu.Unlock() } + +func (f *finalSnap) final() (int, bool) { f.mu.Lock(); defer f.mu.Unlock(); return f.since, f.sent } + +func (f *finalSnap) Digest(d protocol.Digest) error { return f.link.Digest(d) } + +func (f *finalSnap) Chunk(c protocol.Chunk) error { + if c.Final { + f.mu.Lock() + f.since, f.sent = f.tee.count(c.Digest), true + f.mu.Unlock() + } + return f.link.Chunk(c) +} diff --git a/free/ci/internal/cache/remote/restore_test.go b/free/ci/internal/cache/remote/restore_test.go new file mode 100644 index 000000000..156c2753f --- /dev/null +++ b/free/ci/internal/cache/remote/restore_test.go @@ -0,0 +1,223 @@ +package remote + +import ( + "bytes" + "context" + "encoding/base64" + "errors" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/nself-org/plugins/free/ci/internal/blob/transfer" + "github.com/nself-org/plugins/free/ci/internal/cache/entries" + "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/store" + "github.com/nself-org/plugins/free/ci/internal/transport" +) + +// wire joins a coordinator Restore to a NodeRestorer in memory, frame for frame. +type wire struct { + frames chan transport.BlobFrame + offs chan protocol.Offset + chunks atomic.Int32 + // flip, when set, may change a chunk before the node sees it. + flip func(*protocol.Chunk) +} + +func newWire() *wire { + return &wire{frames: make(chan transport.BlobFrame, 8), offs: make(chan protocol.Offset, 64)} +} + +type wireLink struct{ w *wire } + +func (l wireLink) Digest(d protocol.Digest) error { + select { + case l.w.frames <- transport.BlobFrame{Digest: &d}: + return nil + case <-time.After(5 * time.Second): + return errors.New("wire: node not reading") + } +} + +func (l wireLink) Chunk(c protocol.Chunk) error { + l.w.chunks.Add(1) + if l.w.flip != nil { + l.w.flip(&c) + } + select { + case l.w.frames <- transport.BlobFrame{Chunk: &c}: + return nil + case <-time.After(5 * time.Second): + return errors.New("wire: node not reading") + } +} +func (l wireLink) Offsets() <-chan protocol.Offset { return l.w.offs } + +type wireFeed struct{ w *wire } + +func (f wireFeed) Frames() <-chan transport.BlobFrame { return f.w.frames } +func (f wireFeed) Offset(o protocol.Offset) error { f.w.offs <- o; return nil } +func (f wireFeed) Close() {} + +// restoreFixture stores one blob on the coordinator and plans it as a restore. +func restoreFixture(t *testing.T) (*Lease, *store.Store, []byte, protocol.CacheBlob) { + t.Helper() + l, st, _ := coordinatorFixture(t) + data := bytes.Repeat([]byte("restore me "), 110000) // 1.2 MiB: three chunks + sha, size, err := l.h.Blobs.Put(context.Background(), cacheRealm(l.b.Plan.Namespace), bytes.NewReader(data)) + if err != nil { + t.Fatal(err) + } + b := protocol.CacheBlob{Path: "mod", SHA256: sha, Size: size} + l.b.Plan.Restore = []protocol.CacheRestore{{Spec: "go", Key: strings.Repeat("c", 64), Dir: "go", Blobs: []protocol.CacheBlob{b}}} + l.b.Node = "node-b" + l.b.RestoreWait = 5 * time.Second + return l, st, data, b +} + +// runRestore pushes the lease's restore blobs to a node holding copies. +func runRestore(t *testing.T, l *Lease, copies *ReadThrough, w *wire) (map[string]bool, error) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- l.Restore(ctx, wireLink{w}) }() + n := &NodeRestorer{Copies: copies, Feed: wireFeed{w}, Realm: l.b.Plan.Namespace, Wait: 5 * time.Second} + got := n.Fetch(ctx, l.b.Plan.Restore[0].Blobs) + return got, <-done +} + +func TestRestoreColdThenHeldCopyNeedsNoChunks(t *testing.T) { + l, _, data, b := restoreFixture(t) + copies, err := NewReadThrough(t.TempDir(), 8<<20) + if err != nil { + t.Fatal(err) + } + w := newWire() + if got, err := runRestore(t, l, copies, w); err != nil || !got[b.SHA256] { + t.Fatalf("cold restore got=%v err=%v", got, err) + } + if w.chunks.Load() != 3 { + t.Fatalf("cold restore sent %d chunks", w.chunks.Load()) + } + f, err := copies.Open(context.Background(), l.b.Plan.Namespace, b.SHA256, b.Size, nil) + if err != nil { + t.Fatal(err) + } + _ = f.Close() + rows, err := l.h.Ledger.CacheLocality(context.Background(), "node-b", l.b.Plan.Namespace, strings.Repeat("c", 64)) + if err != nil || len(rows) != 1 || rows[0].Bytes != int64(len(data)) { + t.Fatalf("locality=%+v err=%v", rows, err) + } + w2 := newWire() + if got, err := runRestore(t, l, copies, w2); err != nil || !got[b.SHA256] { + t.Fatalf("held restore got=%v err=%v", got, err) + } + if n := w2.chunks.Load(); n != 0 { + t.Fatalf("a held copy cost %d chunk frames", n) + } +} + +func TestRestoreCorruptCopyDiscardedAndFetchedAgain(t *testing.T) { + l, _, _, b := restoreFixture(t) + copies, _ := NewReadThrough(t.TempDir(), 8<<20) + if got, err := runRestore(t, l, copies, newWire()); err != nil || !got[b.SHA256] { + t.Fatalf("seed got=%v err=%v", got, err) + } + name, err := copies.name(l.b.Plan.Namespace, b.SHA256) + if err != nil { + t.Fatal(err) + } + raw, _ := os.ReadFile(name) + raw[7] ^= 0xff + if err = os.WriteFile(name, raw, 0o600); err != nil { + t.Fatal(err) + } + w := newWire() + if got, err := runRestore(t, l, copies, w); err != nil || !got[b.SHA256] { + t.Fatalf("corrupt copy got=%v err=%v", got, err) + } + if w.chunks.Load() != 3 { + t.Fatalf("a corrupt copy was served: %d chunk frames", w.chunks.Load()) + } + f, err := copies.Open(context.Background(), l.b.Plan.Namespace, b.SHA256, b.Size, nil) + if err != nil { + t.Fatalf("copy not replaced: %v", err) + } + _ = f.Close() +} + +// flipOnce changes one byte of the final chunk of the first pass only. +func flipOnce() func(*protocol.Chunk) { + var done atomic.Bool + return func(c *protocol.Chunk) { + if !c.Final || done.Swap(true) { + return + } + raw, _ := base64.StdEncoding.DecodeString(c.Data) + raw[len(raw)/2] ^= 1 + c.Data = base64.StdEncoding.EncodeToString(raw) + } +} + +func TestRestoreFlippedByteFailsAndResumes(t *testing.T) { + l, _, _, b := restoreFixture(t) + copies, _ := NewReadThrough(t.TempDir(), 8<<20) + w := newWire() + w.flip = flipOnce() + if got, err := runRestore(t, l, copies, w); err != nil || !got[b.SHA256] { + t.Fatalf("got=%v err=%v", got, err) + } + if n := w.chunks.Load(); n != 6 { + t.Fatalf("want two full passes (6 chunks), got %d", n) + } + f, err := copies.Open(context.Background(), l.b.Plan.Namespace, b.SHA256, b.Size, nil) + if err != nil { + t.Fatal(err) + } + _ = f.Close() +} + +func TestRestoreAlwaysFlippedLeavesNoCopy(t *testing.T) { + l, _, _, b := restoreFixture(t) + copies, _ := NewReadThrough(t.TempDir(), 8<<20) + w := newWire() + w.flip = func(c *protocol.Chunk) { + if c.Final { + raw, _ := base64.StdEncoding.DecodeString(c.Data) + raw[0] ^= 1 + c.Data = base64.StdEncoding.EncodeToString(raw) + } + } + got, err := runRestore(t, l, copies, w) + if got[b.SHA256] || err == nil { + t.Fatalf("a corrupt transfer was accepted: got=%v err=%v", got, err) + } + if _, err = copies.Open(context.Background(), l.b.Plan.Namespace, b.SHA256, b.Size, nil); err == nil { + t.Fatal("a failed fetch left a copy to bind") + } +} + +func TestRestoreRefusalsAreAudited(t *testing.T) { + l, s, _, b := restoreFixture(t) + ctx := context.Background() + var refused *entries.Refused + if _, _, err := l.OpenRestore(ctx, strings.Repeat("d", 64)); !errors.As(err, &refused) || refused.Reason != entries.ReasonNotPlanned { + t.Fatalf("outside the plan: %v", err) + } + assertAudit(t, s, entries.ReasonNotPlanned) + cctx, cancel := context.WithCancel(ctx) + cancel() + if _, _, err := l.OpenRestore(cctx, b.SHA256); err == nil { + t.Fatal("a cancelled request was served") + } + assertAudit(t, s, "cancelled") + l.b.Epoch++ // a request from an epoch that is no longer the lease's + if _, _, err := l.OpenRestore(ctx, b.SHA256); !transfer.IsReason(err, transfer.ReasonStaleEpoch) { + t.Fatalf("stale epoch: %v", err) + } + assertAudit(t, s, transfer.ReasonStaleEpoch) +} diff --git a/free/ci/internal/run/remote_cache_hook.go b/free/ci/internal/run/remote_cache_hook.go index 202237fb8..fdf7a493d 100644 --- a/free/ci/internal/run/remote_cache_hook.go +++ b/free/ci/internal/run/remote_cache_hook.go @@ -264,3 +264,13 @@ func (s cacheLeaseSink) Result(ctx context.Context, r protocol.Result) error { } return s.Sink.Result(ctx, r) } + +// Restore pushes the lease's planned restore blobs to the node (carrier A). The +// lease audits its own refusals; a failed push leaves the node a cold miss. +func (s cacheLeaseSink) Restore(ctx context.Context, link transport.RestoreLink) error { + if err := s.l.Restore(ctx, link); err != nil { + cacheOn.warn("remote cache restore: " + firstLineOf(err.Error())) + return err + } + return nil +} diff --git a/free/ci/internal/transport/restore.go b/free/ci/internal/transport/restore.go new file mode 100644 index 000000000..7ef21e61a --- /dev/null +++ b/free/ci/internal/transport/restore.go @@ -0,0 +1,49 @@ +package transport + +// Purpose: the seams of the reverse blob path (Epic CACHE D7, P7-CACHE-21): the +// coordinator pushes the cache blobs a lease restores to the node, and the node +// session hands those frames to the job hook that is receiving them. +// Inputs: a Sink that can open a lease's planned restore blobs (Restorer), and +// the digest and chunk frames a node session receives (BlobFeed). +// Outputs: RestoreLink for a driver to give the Sink, BlobFrame for the node. +// Constraints: a carrier that cannot send coordinator frames to a node never +// calls Restore, and its node session never offers a feed, so a cold restore +// degrades at once instead of waiting for frames that cannot arrive. + +import ( + "context" + + "github.com/nself-org/plugins/free/ci/internal/protocol" +) + +// RestoreLink is the coordinator's end of one lease's reverse blob path, bound +// to the lease and epoch of the session (a frame it sends carries both). +type RestoreLink interface { + Digest(protocol.Digest) error + Chunk(protocol.Chunk) error + // Offsets carries the node's offset answers for this lease. + Offsets() <-chan protocol.Offset +} + +// Restorer is implemented by a Sink that owns the cache plan of its lease. A +// driver calls Restore once, after the lease frame is sent, on its own goroutine; +// the error is the Sink's to audit, the job never waits for it. +type Restorer interface { + Restore(ctx context.Context, link RestoreLink) error +} + +// BlobFrame is one coordinator-to-node blob frame: exactly one field is set. +type BlobFrame struct { + Digest *protocol.Digest + Chunk *protocol.Chunk +} + +// BlobFeed is the node session's end of a lease's reverse blob path. +type BlobFeed interface { + // Frames carries the lease's digest and chunk frames in arrival order. + Frames() <-chan BlobFrame + // Offset answers the coordinator. + Offset(protocol.Offset) error + // Close tells the session nobody reads Frames any more; later frames are dropped. + Close() +} diff --git a/free/ci/internal/transport/ssh/restore.go b/free/ci/internal/transport/ssh/restore.go new file mode 100644 index 000000000..5e1a03d24 --- /dev/null +++ b/free/ci/internal/transport/ssh/restore.go @@ -0,0 +1,60 @@ +package ssh + +// Purpose: the coordinator end of the reverse blob path on carrier A (P7-CACHE-21, +// Epic CACHE D7): after the lease is sent, a Sink that owns the lease's cache plan +// (transport.Restorer) pushes the restore blobs as digest and chunk frames, and the +// node's offset answers are routed back to it. +// Inputs: the node's `offset` frames for this lease and epoch. +// Outputs: `digest` and `chunk` frames fenced to the lease and epoch. +// Constraints: the push runs beside the serve loop and ends with it (the context is +// cancelled when the lease ends or is cancelled); an offset for another lease or +// epoch is dropped; the answer channel never blocks the read loop (a full channel +// drops the oldest answer, the newest one is the node's current position). + +import ( + "context" + + "github.com/nself-org/plugins/free/ci/internal/protocol" + "github.com/nself-org/plugins/free/ci/internal/transport" +) + +// offsetBuffer is how many unread node answers a session keeps. +const offsetBuffer = 64 + +type restoreLink struct{ s *sess } + +func (l restoreLink) Digest(d protocol.Digest) error { return l.s.send("digest", d, true) } +func (l restoreLink) Chunk(c protocol.Chunk) error { return l.s.send("chunk", c, true) } +func (l restoreLink) Offsets() <-chan protocol.Offset { return l.s.offsets } + +var _ transport.RestoreLink = restoreLink{} + +// startRestore begins the push when the Sink has restore blobs; the returned func stops it. +func (s *sess) startRestore(ctx context.Context) func() { + r, ok := s.sink.(transport.Restorer) + if !ok { + return func() {} + } + rctx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + _ = r.Restore(rctx, restoreLink{s}) // the Sink audits a refusal; the node runs cold + }() + return func() { cancel(); <-done } +} + +// onOffset hands a node answer to the push. +func (s *sess) onOffset(o protocol.Offset) { + for { + select { + case s.offsets <- o: + return + default: + } + select { + case <-s.offsets: + default: + } + } +} diff --git a/free/ci/internal/transport/ssh/run.go b/free/ci/internal/transport/ssh/run.go index 427dc1d58..4407b633c 100644 --- a/free/ci/internal/transport/ssh/run.go +++ b/free/ci/internal/transport/ssh/run.go @@ -45,6 +45,7 @@ type sess struct { seq uint64 frames chan frame + offsets chan protocol.Offset // the node's offset answers for the restore push (restore.go) stop chan struct{} readerDone chan struct{} errDone <-chan struct{} @@ -85,7 +86,7 @@ func (d *Driver) Run(ctx context.Context, node model.Capability, lease protocol. if err != nil { return newErr(CodeTransport, ReasonConnect, "cannot start ssh for node %s: %v", node.Identity.ID, err) } - s := &sess{d: d, node: node, lease: lease, sink: sink, conn: conn, frames: make(chan frame, 64), + s := &sess{d: d, node: node, lease: lease, sink: sink, conn: conn, frames: make(chan frame, 64), offsets: make(chan protocol.Offset, offsetBuffer), stop: make(chan struct{}), readerDone: make(chan struct{}), origin: time.Now(), blobs: map[string]*blobState{}, committed: map[string]bool{}} s.stderr, s.errDone = drain(conn.Stderr) defer s.close() diff --git a/free/ci/internal/transport/ssh/serve.go b/free/ci/internal/transport/ssh/serve.go index 5ab1f76a6..56be42fff 100644 --- a/free/ci/internal/transport/ssh/serve.go +++ b/free/ci/internal/transport/ssh/serve.go @@ -27,6 +27,7 @@ func (s *sess) serve(ctx context.Context, bundle string) error { if err := s.sendSecrets(ctx); err != nil { return err } + defer s.startRestore(ctx)() uploaded := make(chan error, 1) if bundle != "" { go func() { uploaded <- s.upload(ctx, bundle) }() @@ -236,6 +237,10 @@ func (s *sess) handle(ctx context.Context, m protocol.Message) (done bool, err e return false, nil } return false, s.onChunk(*b) + case *protocol.Offset: + if s.scoped(m) { + s.onOffset(*b) + } } return false, nil } From 6d90aea3960445d29469f891ff78f03c02a7e83e Mon Sep 17 00:00:00 2001 From: Aric Camarata Date: Mon, 26 Oct 2026 15:38:29 -0400 Subject: [PATCH 2/2] fix(ci): restore review fixes: attempt_state audit reason, digest normalising, rename carrier B test --- free/ci/internal/cache/remote/integration_restore_test.go | 4 +++- free/ci/internal/cache/remote/node_restore.go | 8 +++++--- free/ci/internal/cache/remote/restore_coord.go | 3 ++- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/free/ci/internal/cache/remote/integration_restore_test.go b/free/ci/internal/cache/remote/integration_restore_test.go index 08b3f3a7b..74625a52f 100644 --- a/free/ci/internal/cache/remote/integration_restore_test.go +++ b/free/ci/internal/cache/remote/integration_restore_test.go @@ -222,7 +222,9 @@ func TestRemoteCacheRestoreCorrupt(t *testing.T) { } } -func TestRemoteCacheRestoreAgent(t *testing.T) { +// TODO: a real carrier-B restore test must replace this once agentapi can carry the +// reverse blob frames; until then this pins only the degrade path. +func TestRemoteCacheRestoreAgentDegradesToColdMiss(t *testing.T) { data := carrierCacheArchive(t) node := newRestoreNode(t, false) r := newRig(t) diff --git a/free/ci/internal/cache/remote/node_restore.go b/free/ci/internal/cache/remote/node_restore.go index 719048542..0736e7f93 100644 --- a/free/ci/internal/cache/remote/node_restore.go +++ b/free/ci/internal/cache/remote/node_restore.go @@ -61,7 +61,7 @@ func (n *NodeRestorer) Fetch(ctx context.Context, blobs []protocol.CacheBlob) ma if fr.Digest == nil { continue // a chunk with no open blob } - sha := strings.ToLower(strings.TrimPrefix(fr.Digest.SHA256, "sha256:")) + sha := normDigest(fr.Digest.SHA256) b, ok := pending[sha] if !ok || fr.Digest.Stream != StreamCache || fr.Digest.Size != b.Size { continue // outside the plan: never answered, never stored @@ -103,6 +103,8 @@ func (n *NodeRestorer) next(ctx context.Context) (transport.BlobFrame, error) { } } +func normDigest(d string) string { return strings.ToLower(strings.TrimPrefix(d, "sha256:")) } + func (n *NodeRestorer) answer(b protocol.CacheBlob, off int64) error { return n.Feed.Offset(protocol.Offset{Stream: StreamCache, SHA256: b.SHA256, Offset: off}) } @@ -134,14 +136,14 @@ func (n *NodeRestorer) stream(ctx context.Context, w io.Writer, b protocol.Cache return err } if fr.Digest != nil { - if strings.TrimPrefix(fr.Digest.SHA256, "sha256:") == b.SHA256 { + if normDigest(fr.Digest.SHA256) == b.SHA256 { n.carry = &fr return errRestart } continue } c := fr.Chunk - if c == nil || c.Stream != StreamCache || strings.TrimPrefix(c.Digest, "sha256:") != b.SHA256 { + if c == nil || c.Stream != StreamCache || normDigest(c.Digest) != b.SHA256 { continue } if c.Offset != next { diff --git a/free/ci/internal/cache/remote/restore_coord.go b/free/ci/internal/cache/remote/restore_coord.go index 7b0daa9fb..3f410707e 100644 --- a/free/ci/internal/cache/remote/restore_coord.go +++ b/free/ci/internal/cache/remote/restore_coord.go @@ -67,7 +67,7 @@ func (l *Lease) OpenRestore(ctx context.Context, sha string) (io.ReadCloser, pro return nil, b, err } if state != "leased" && state != "running" { - return nil, b, l.refuse(ctx, transfer.ReasonStaleEpoch) + return nil, b, l.refuse(ctx, entries.ReasonAttemptState) } realm := cacheRealm(l.b.Plan.Namespace) info, err := l.h.Blobs.Stat(realm, sha) @@ -109,6 +109,7 @@ func (l *Lease) Restore(ctx context.Context, link transport.RestoreLink) error { done[b.SHA256] = true } if l.b.Node != "" { + // Realm is the wire namespace, as finalize.go writes it at commit (not cas.CacheRealm). row := store.CacheLocalityRow{Node: l.b.Node, Realm: l.b.Plan.Namespace, Key: r.Key, Bytes: total} if err := l.h.Ledger.CacheLocalityPut(ctx, row); err != nil && l.h.Warn != nil { l.h.Warn(fmt.Errorf("cache locality: %w", err))