Continue the chat a person picks; keep the retry card; honest waits on a slow relay; trust-proxy for every relay limit

cmd/codeaf/synctake.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"log"
+	"net/url"
 	"os"
 	"path/filepath"
 	"time"
@@ -372,3 +373,18 @@ func (t takeover) SafeToContinue(ctx context.Context, id, current string, automa
 	}
 	return !dirty, err
 }
+
+// ReleaseUnused leaves the installed files and kept work intact, just as closing
+// a chat does, while returning an abandoned automatic acquisition to the directory.
+func (t takeover) ReleaseUnused(ctx context.Context, id string, fence uint64) error {
+	return t.c.ReleaseUnused(ctx, id, fence)
+}
+
+// RelayHost names the endpoint without exposing a path or credentials in chat copy.
+func (s surfaceMachines) RelayHost() string {
+	parsed, err := url.Parse(s.Relay)
+	if err != nil {
+		return ""
+	}
+	return parsed.Hostname()
+}

docs/changes/unreleased/2089-relay-l3-findings.md

@@ -0,0 +1,19 @@
+---
+kind: fixed
+title: Continuing opens the chat you picked, a refused continue keeps its card, and a slow relay says so
+pr: 2089
+surface: [chat, remote, docs]
+invalidates:
+  - "A bare codeaf chat that was continuing the newest chat another computer let go of answered every explicit continue on Home with a chat is already continuing here, and the older chat opened; now a confirmed different chat waits, the automatic move finishes and is let go without opening, and the chosen chat opens. The busy line now answers only a second pick during a continue the person started."
+  - "A continue refused because the chat's holder changed showed another device continued this chat first in the footer only; every such refusal now keeps the 1 retry / 2 leave it there card and the draft, and when the holder named on the card simply reopened the chat the card asks again instead of blaming another device."
+  - "Home gave up on the chat listing after 5 seconds and said other machines did not answer in time; it now waits up to 25 seconds, says waiting for <relay host>; still checking after 5, keeps its rows, and still shows a real outage at once."
+  - "The first message sent right after a move could be left in the box under Home's timeout sentence; a send right after this window's own move now waits up to 30 seconds for the arriving chat to be ready and sends once, and a send that cannot leave says the message is still in the box and names the relay."
+  - "A self-hosted codeaf relay counted the first X-Forwarded-For address for its registration and dial limits whenever a request sent one, so a relay facing the internet could be given any address; forwarding headers now count only with --trust-proxy, for registration, dial and pairing limits alike, and a relay behind a proxy must pass it or every caller shares one allowance."
+  - "A chat this computer had never held came over serial object batches whenever the shared store had exchanged any other chat with the relay; it now downloads its frames, and a chat this computer already holds still fetches only what changed."
+---
+
+Found by the two-network test of the relay (lane L3). Each continue also writes
+one `chat.log` line with the time spent in each phase, so a slow move can be read
+instead of guessed. Consent to continue still binds the holder and its ownership
+number shown on the card; nothing about ownership rules, wire or stored formats,
+or the hosted relay changed.

docs/relay.md

@@ -19,12 +19,14 @@ The separate `relay` binary accepts the same flags.
 | `--store <dir>` | Persistent coordination and encrypted chunks. Without it, only the blind transport and pairing mailbox run; chat sync needs storage. |
 | `--quiet` | Suppress arrival, departure and request logs. |
 | `--status=false` | Disable `GET /status`. |
-| `--trust-proxy` | Use the first `X-Forwarded-For` address for pairing limits, only behind a proxy that sets it. |
+| `--trust-proxy` | use the first `X-Forwarded-For` address for registration, dial and pairing limits; absent or blank falls back to the socket peer. only behind a proxy that sets it. |
 
 Set `CODEAF_SYNC_URL=https://relay.example.com` on each computer, or use
 `codeaf pair --via https://relay.example.com`. Pair the computers with the same
 identity and relay. `CODEAF_SYNC_URL=off` stops sync, including existing clients.
 
+forwarding headers are ignored by default. behind Railway, a load balancer or nginx, pass `--trust-proxy` or every caller shares one allowance. enable it only behind a proxy that sets the header.
+
 The relay supplies no TLS or list of allowed people. Use a TLS proxy and a private
 network or a proxy that checks callers: any valid signature can create an identity.
 Logs contain a short identity id, route pattern, status and byte counts, never a body.

internal/cellstore/chat_warmth_test.go

@@ -0,0 +1,29 @@
+package cellstore
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/Agent-Field/codeaf/internal/cell"
+)
+
+func TestChatWarmthBelongsToItsOwnHead(t *testing.T) {
+	t.Setenv("HOME", t.TempDir())
+	t.Setenv("CODEAF_TELEMETRY", "off")
+	data := t.TempDir()
+	c := cell.Cell{ID: "new-chat", Root: t.TempDir()}
+	e := SyncEngine{Ledger: "used", Target: func(cell.Cell) Target { return Target{DataDir: data} }}
+	if err := os.WriteFile(filepath.Join(data, "published.used"), []byte("earlier chat's objects"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	if e.Holds(c) {
+		t.Fatal("an unrelated chat's ledger made this chat warm")
+	}
+	if err := noteAdopted(c, "held-head"); err != nil {
+		t.Fatal(err)
+	}
+	if !e.Holds(c) {
+		t.Fatal("this chat's adopted history was forgotten")
+	}
+}

internal/cellstore/syncops.go

@@ -241,12 +241,15 @@ func (e SyncEngine) importWith(ctx context.Context, c cell.Cell, op importOp) (i
 	return out.Imported, e.ask(ctx, c, op, &out)
 }
 
-// Holds implements the sync seam: the ledger file the engine keeps beside the
-// store is written by every export and import against this relay, so it has
-// content exactly when the device has exchanged objects with the relay.
+// Holds reports history this computer held for this chat. The publication
+// ledger belongs to the shared store, so another chat's exchange proves nothing
+// about this one. Local adoption and turn records already supply this evidence.
 func (e SyncEngine) Holds(c cell.Cell) bool {
-	info, err := os.Stat(filepath.Join(e.Target(c).DataDir, "published."+e.Ledger))
-	return err == nil && info.Size() > 0
+	if _, ok := readAdoption(c); ok {
+		return true
+	}
+	turn, err := lastTurn(rel(c, TurnsPath))
+	return err == nil && turn != nil
 }
 
 // Materialize implements the sync seam.

internal/cellsync/chat_priming_test.go

@@ -0,0 +1,210 @@
+package cellsync
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+
+	"github.com/Agent-Field/codeaf/internal/blobstore"
+	"github.com/Agent-Field/codeaf/internal/cell"
+	"github.com/Agent-Field/codeaf/internal/cell/celltest"
+	"github.com/Agent-Field/codeaf/internal/cellstore"
+	"github.com/Agent-Field/codeaf/internal/furrow"
+)
+
+type chatFetchCounts struct {
+	blobstore.Store
+	frames, objects, bytes atomic.Int64
+}
+
+func (s *chatFetchCounts) GetFrame(ctx context.Context, id string) ([]byte, error) {
+	s.frames.Add(1)
+	raw, err := s.Store.GetFrame(ctx, id)
+	s.bytes.Add(int64(len(raw)))
+	return raw, err
+}
+func (s *chatFetchCounts) GetMany(ctx context.Context, ids []string) ([]blobstore.Object, error) {
+	s.objects.Add(1)
+	objects, err := s.Store.GetMany(ctx, ids)
+	for _, o := range objects {
+		s.bytes.Add(int64(len(o.Bytes)))
+	}
+	return objects, err
+}
+
+type wantCountingEngine struct {
+	Engine
+	wants atomic.Int64
+}
+
+func (e *wantCountingEngine) Want(ctx context.Context, c cell.Cell, h string) ([]string, error) {
+	e.wants.Add(1)
+	return e.Engine.Want(ctx, c, h)
+}
+
+type usedStoreRig struct {
+	sender      cellstore.Engine
+	pub         *Publisher
+	receiver    cellstore.SyncEngine
+	fetch       *Fetcher
+	counts      *chatFetchCounts
+	cell        cell.Cell
+	source      cell.Cell
+	earlier     cell.Cell
+	earlierHead string
+	body        []byte
+	head        string
+}
+
+func newUsedStoreRig(t *testing.T) *usedStoreRig {
+	t.Helper()
+	t.Setenv("HOME", t.TempDir())
+	t.Setenv("CODEAF_TELEMETRY", "off")
+	celltest.RequireEngine(t)
+	binary, err := furrow.ResolveOwned()
+	if err != nil {
+		t.Skip(err)
+	}
+	t.Cleanup(cellstore.StopSpawned)
+	sender := cellstore.Engine{Binary: binary, DataRoot: t.TempDir(), Transport: cellstore.Spawn{Binary: binary}}
+	receiver := cellstore.Engine{Binary: binary, DataRoot: t.TempDir(), Transport: cellstore.Spawn{Binary: binary}}
+	keys := cellstore.SyncKeys{CellKey: make([]byte, 32), Dedup: make([]byte, 32)}
+	send := sender.Sync(keys, "0123456789abcdef")
+	recv := receiver.Sync(keys, "0123456789abcdef")
+	r := newRig(t)
+	pub := &Publisher{Engine: send, Store: r.mem, Dir: r.dirA}
+	counts := &chatFetchCounts{Store: r.mem}
+	f := &Fetcher{Engine: recv, Store: counts, Dir: r.dir.For(devB), Inbox: recv.Inbox}
+	makeChat := func() (cell.Cell, cell.Cell) {
+		c, err := cell.CreateIn(t.TempDir(), cell.Options{Class: cell.Sandboxed})
+		if err != nil {
+			t.Fatal(err)
+		}
+		local := cell.Cell{ID: c.ID, Root: filepath.Join(t.TempDir(), c.ID)}
+		if err := os.MkdirAll(local.Root, 0700); err != nil {
+			t.Fatal(err)
+		}
+		return c, local
+	}
+	publish := func(c cell.Cell, body []byte) string {
+		if err := os.WriteFile(filepath.Join(c.Root, "payload"), body, 0600); err != nil {
+			t.Fatal(err)
+		}
+		sealed, err := sender.Seal(t.Context(), c, cellstore.TurnInfo{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := cellstore.Await(t.Context(), c); err != nil {
+			t.Fatal(err)
+		}
+		if err := pub.Publish(t.Context(), &Driving{Cell: c}, sealed.Turn.ID, PublishInfo{Class: "chat"}); err != nil {
+			t.Fatal(err)
+		}
+		return sealed.Turn.ID
+	}
+	earlier, localEarlier := makeChat()
+	first := publish(earlier, []byte("an earlier chat"))
+	if err := f.Fetch(t.Context(), localEarlier, first); err != nil {
+		t.Fatal(err)
+	}
+	source, local := makeChat()
+	body := make([]byte, 2<<20)
+	if _, err := rand.Read(body); err != nil {
+		t.Fatal(err)
+	}
+	head := publish(source, body)
+	counts.frames.Store(0)
+	counts.objects.Store(0)
+	counts.bytes.Store(0)
+	return &usedStoreRig{sender: sender, pub: pub, receiver: recv, fetch: f, counts: counts, cell: local, source: source, earlier: earlier, earlierHead: first, body: body, head: head}
+}
+func TestANewChatInAUsedStorePrimesFromItsPlan(t *testing.T) {
+	r := newUsedStoreRig(t)
+	// Skipping the plan was the old used-store path. Fetch the very same source
+	// through that object-only door in another already-used store for comparison.
+	local := cellstore.Engine{Binary: r.sender.Binary, DataRoot: t.TempDir(), Transport: cellstore.Spawn{Binary: r.sender.Binary}}
+	engine := local.Sync(r.receiver.Keys, r.receiver.Ledger)
+	before := &chatFetchCounts{Store: r.counts.Store}
+	plain := &Fetcher{Engine: engine, Store: before, Inbox: engine.Inbox}
+	if err := plain.Fetch(t.Context(), cell.Cell{ID: r.earlier.ID, Root: t.TempDir()}, r.earlierHead); err != nil {
+		t.Fatal(err)
+	}
+	before.frames.Store(0)
+	before.objects.Store(0)
+	before.bytes.Store(0)
+	plainCell := cell.Cell{ID: r.cell.ID, Root: t.TempDir()}
+	if err := plain.Fetch(t.Context(), plainCell, r.head); err != nil {
+		t.Fatal(err)
+	}
+	original, err := os.ReadFile(filepath.Join(plainCell.Root, "payload"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(original, r.body) || before.frames.Load() != 0 || before.objects.Load() == 0 {
+		t.Fatal("object-only baseline did not complete the same bytes")
+	}
+	t.Logf("object-only cold baseline in used store: frame GETs=%d object POSTs=%d bytes=%d", before.frames.Load(), before.objects.Load(), before.bytes.Load())
+	counted := &wantCountingEngine{Engine: r.receiver}
+	r.fetch.Engine = counted
+	if err := r.fetch.Fetch(t.Context(), r.cell, r.head); err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("cold chat in used store: frame GETs=%d object POSTs=%d bytes=%d", r.counts.frames.Load(), r.counts.objects.Load(), r.counts.bytes.Load())
+	if r.counts.frames.Load() == 0 || r.counts.objects.Load() != 0 {
+		t.Fatal("new chat skipped its frame plan")
+	}
+	if counted.wants.Load() < 2 {
+		t.Fatal("priming skipped the completeness loop")
+	}
+	got, err := os.ReadFile(filepath.Join(r.cell.Root, "payload"))
+	if err != nil || !bytes.Equal(got, r.body) || !bytes.Equal(got, original) {
+		t.Fatal("received different bytes", err)
+	}
+}
+func TestAWarmChatStillFetchesOnlyItsDelta(t *testing.T) {
+	r := newUsedStoreRig(t)
+	if err := r.fetch.Fetch(t.Context(), r.cell, r.head); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(r.source.Root, "small"), []byte("one file changed"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	sealed, err := r.sender.Seal(t.Context(), r.source, cellstore.TurnInfo{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := cellstore.Await(t.Context(), r.source); err != nil {
+		t.Fatal(err)
+	}
+	// Reuse the actual published head/ownership, just as the existing writer does.
+	view, err := r.pub.Dir.Cell(t.Context(), r.source.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	drv := &Driving{Cell: r.source, Fence: view.Cell.Lease.Fence, Head: view.Cell.Head, Frames: view.Cell.Frames}
+	if err := r.pub.Publish(t.Context(), drv, sealed.Turn.ID, PublishInfo{Class: "chat"}); err != nil {
+		t.Fatal(err)
+	}
+	r.counts.frames.Store(0)
+	r.counts.objects.Store(0)
+	r.counts.bytes.Store(0)
+	if err := r.fetch.Fetch(t.Context(), r.cell, sealed.Turn.ID); err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("warm delta: frame GETs=%d object POSTs=%d bytes=%d", r.counts.frames.Load(), r.counts.objects.Load(), r.counts.bytes.Load())
+	if r.counts.frames.Load() != 0 || r.counts.objects.Load() == 0 || r.counts.bytes.Load() >= int64(len(r.body)) {
+		t.Fatal("warm chat redownloaded its history")
+	}
+	got, err := os.ReadFile(filepath.Join(r.cell.Root, "payload"))
+	if err != nil || !bytes.Equal(got, r.body) {
+		t.Fatal("warm history changed", err)
+	}
+	small, err := os.ReadFile(filepath.Join(r.cell.Root, "small"))
+	if err != nil || string(small) != "one file changed" {
+		t.Fatal("delta missing", err)
+	}
+}

internal/cellsync/engine.go

@@ -33,9 +33,9 @@ type Engine interface {
 	// reach so far, and leaves every other file, which may be waiting for a
 	// frame that has not landed. A take calls it while frames still download.
 	ImportPartial(ctx context.Context, c cell.Cell, head, inbox string) (int, error)
-	// Holds reports whether this device already holds objects of c that the
-	// relay holds too: its ledger for this relay is not empty. A device that
-	// does has most of the chat's history and wants only a delta (contract
+	// Holds reports whether this device has already held this chat's history.
+	// Activity for another chat in a shared store says nothing about this one.
+	// A warm chat wants only its delta (contract
 	// §22.5), so priming it with the chat's whole frame plan would download
 	// what it has.
 	Holds(c cell.Cell) bool

internal/cellsync/fetch.go

@@ -137,8 +137,7 @@ func (f *Fetcher) prime(ctx context.Context, c cell.Cell, head, inbox string) er
 	}
 	inbox = owned
 	defer os.RemoveAll(inbox)
-	// A device that already exchanged objects with this relay holds most of the
-	// chat: the plan names its whole upload history, so priming it would
+	// A device that already held this chat has most of its history: the plan names its whole upload history, so priming it would
 	// download what it has to learn about a one-file change. The want loop
 	// fetches a warm take's delta by rid.
 	if f.Engine.Holds(c) {

internal/handoff/handoff.go

@@ -107,7 +107,9 @@ func (t Taker) Take(ctx context.Context, id string) (Taken, error) {
 // TakeSeen continues the lease the person saw on the card. Older callers
 // without that reading still bind the claim to the record read before fetch.
 func (t Taker) TakeSeen(ctx context.Context, id string, seen *directory.LeaseExpectation) (Taken, error) {
+	directoryRead := Trace(t.OnPhase, "directory")
 	view, err := t.Dir.Cell(ctx, id)
+	directoryRead()
 	if err != nil {
 		return Taken{}, fmt.Errorf("handoff: read chat %s: %w", id, err)
 	}
@@ -266,7 +268,9 @@ func (t Taker) fetchAndAcquire(ctx context.Context, stage cell.Cell, from, head
 	if t.OnPhase != nil {
 		t.OnPhase(Phase{Name: "first acquire", Started: time.Now()})
 	}
+	acquiring := Trace(t.OnPhase, "acquire")
 	got, err := t.Dir.Acquire(ctx, stage.ID, how)
+	acquiring()
 	if err != nil {
 		return "", 0, nil, fmt.Errorf("handoff: take %s: %w", stage.ID, err)
 	}
@@ -400,7 +404,10 @@ func swap(stage, root string) error {
 // here hands it back, so the chat is takeable at once instead of after a TTL.
 func (t Taker) open(ctx context.Context, c cell.Cell, fence uint64) (cell.Cell, error) {
 	for _, hook := range t.After {
-		if err := hook(ctx, c); err != nil {
+		done := Trace(t.OnPhase, "hooks")
+		err := hook(ctx, c)
+		done()
+		if err != nil {
 			return cell.Cell{}, t.giveBack(ctx, c.ID, fence, fmt.Errorf("handoff: %w", err))
 		}
 	}