fix(proxies): retry a copy the cap reset could not start, so it does not sit out a whole period

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **A copy the cap reset could not start stayed stopped for a whole period.** It kept its "stopped by the cap" mark, and nothing tried again until the next reset or a Start by hand. The once-a-minute pass now starts such a copy while the proxy is uncapped, and leaves alone copies you stopped yourself.
 - **Redeploying a proxied copy could leave the proxy with no copy at all.** The old earner and tunnel were removed before the new tunnel started, so a tunnel that could not start (an image missing locally, a proxy that does not answer) cost the running copy. The new tunnel now starts under a staged name next to the running copy, and the old pair goes only after the new tunnel has proved an exit.
 - **The Fleet page could draw over another view, and lose a half-typed proxy.** Fleet reads the backend twice before it draws, so a user who opened another view meanwhile got the Fleet page back. Any proxy action redraws the page, which also reset the open proxy editor and dropped unsaved values, a typed password included. A redraw now gives up when Fleet is no longer shown or a newer redraw started, and keeps the open editor when the same proxy's editor is drawn again.
 - **The Proxies section had no help link.** It now links to the published page on proxied copies, https://geiserx.github.io/CashPilot/proxies/.

docs/proxies.md

@@ -164,8 +164,8 @@ proxy is stopped and you get a notice. Only the copies that were running are
 marked as stopped by the cap, so a copy you had already stopped stays stopped
 after the reset. While the proxy is capped, Start is disabled on its copies:
 reset the usage or raise the cap first. A reset starts every marked copy; one
-whose start fails keeps its mark, the failure shows in its events, and the next
-reset or a Start by hand tries again.
+whose start fails keeps its mark, the failure shows in its events, and every
+minute after that the app tries again while the proxy stays uncapped.
 
 ## Tailscale exit nodes
 

internal/services/proxies.go

@@ -49,6 +49,11 @@ var detectOwnIP = func(ctx context.Context) string {
 // write that one column, and two that overlap would lose one write.
 type proxyState struct {
 	proxyMu sync.Mutex
+	// copyMu holds one mutex per copy ("id/slug"). It orders an owner's Start or Stop,
+	// a reset's start and the tick's retry of the same copy, so a stop that returned
+	// is never undone by a retry that read the mark before it, and two starts never
+	// overlap. Other copies never wait on it.
+	copyMu sync.Map
 	// last is the latest status read, for the UI.
 	lastMu sync.Mutex
 	last   []runtime.ProxiedCopy
@@ -489,6 +494,9 @@ func (m *Manager) otherExits(self int64) []proxies.Exit {
 // StopProxied stops a proxied copy. A stop by hand is the owner's decision, so a
 // later cap reset does not start it again.
 func (m *Manager) StopProxied(ctx context.Context, slug string, id int64) error {
+	mu := m.copyLock(id, slug)
+	mu.Lock()
+	defer mu.Unlock()
 	p, err := m.proxier()
 	if err != nil {
 		return err
@@ -506,6 +514,9 @@ func (m *Manager) StopProxied(ctx context.Context, slug string, id int64) error
 
 // StartProxied starts a proxied copy, refused while its proxy is capped.
 func (m *Manager) StartProxied(ctx context.Context, slug string, id int64) error {
+	mu := m.copyLock(id, slug)
+	mu.Lock()
+	defer mu.Unlock()
 	d, s, err := m.loadProxy(id)
 	if err != nil {
 		return err
@@ -593,11 +604,17 @@ func (m *Manager) releaseCapped(ctx context.Context, id int64, d proxies.Definit
 		if !c.StoppedByCap {
 			continue
 		}
-		if _, err := p.StartProxied(ctx, c.Slug, id, m.stopTimeout(c.Slug), specOf(d), ownIP); err != nil {
+		mu := m.copyLock(id, c.Slug)
+		mu.Lock()
+		_, err := p.StartProxied(ctx, c.Slug, id, m.stopTimeout(c.Slug), specOf(d), ownIP)
+		if err == nil {
+			_ = m.store.SetProxiedCopyCapped(id, c.Slug, false)
+		}
+		mu.Unlock()
+		if err != nil {
 			m.store.RecordEvent(c.Slug, "proxied_start_error", maskSecrets(err.Error(), d))
 			continue
 		}
-		_ = m.store.SetProxiedCopyCapped(id, c.Slug, false)
 		started = append(started, c.Slug)
 	}
 	return started, nil
@@ -707,7 +724,11 @@ func (m *Manager) ProxyTick(ctx context.Context) ([]CapNotice, error) {
 			continue
 		}
 		if reset && next.CappedAt == "" {
-			_, _ = m.releaseCapped(ctx, id, d) // the next reset or a start by hand retries
+			_, _ = m.releaseCapped(ctx, id, d)
+		} else if next.CappedAt == "" {
+			// A reset whose start failed leaves the copy marked as the cap's and down;
+			// while the proxy stays uncapped, every pass tries again.
+			m.restartCapStoppedDown(ctx, p, id, d, mine)
 		}
 		if next.CappedAt == "" {
 			continue
@@ -736,13 +757,108 @@ func (m *Manager) ProxyTick(ctx context.Context) ([]CapNotice, error) {
 	return crossedNow, nil
 }
 
+// restartCapStoppedDown starts this proxy's copies that are still marked as stopped
+// by the cap and are not running. A copy the owner stopped is never marked, so it
+// stays stopped; one that is running is left alone.
+func (m *Manager) restartCapStoppedDown(ctx context.Context, p runtime.ProxyProvider, id int64, d proxies.Definition, status []runtime.ProxiedCopy) {
+	records, err := m.store.ListProxiedCopies(id)
+	if err != nil {
+		return
+	}
+	// Only a running earner counts: a tunnel left up by a start that timed out is not
+	// a copy that runs, so its mark stays and the next pass tries again.
+	running := map[string]bool{}
+	for _, c := range status {
+		running[c.Slug] = c.Status == "running" || c.Status == "restarting"
+	}
+	var ownIP string
+	detected := false // one lookup per pass, even when it fails
+	for _, r := range records {
+		if !r.StoppedByCap {
+			continue
+		}
+		if running[r.Slug] {
+			// Started, but clearing its mark failed: clear it now, or a later crash
+			// of the copy would be read as the cap's stop.
+			if err := m.store.SetProxiedCopyCapped(id, r.Slug, false); err != nil {
+				m.store.RecordEvent(r.Slug, "proxied_mark_error", err.Error())
+			}
+			continue
+		}
+		if !detected {
+			ownIP, detected = detectOwnIP(ctx), true
+		}
+		m.retryCapStart(ctx, p, id, r.Slug, ownIP)
+	}
+}
+
+// copyLock is the mutex of one copy; see proxyState.copyMu.
+func (m *Manager) copyLock(id int64, slug string) *sync.Mutex {
+	mu, _ := m.proxy.copyMu.LoadOrStore(fmt.Sprintf("%d/%s", id, slug), &sync.Mutex{})
+	return mu.(*sync.Mutex)
+}
+
+// retryCapStart starts one cap-stopped copy, unless an owner's Start or Stop, or a
+// reset, got to it first: the mark is read again under the copy's lock, which those
+// hold for their whole run. It reports whether the copy was started.
+func (m *Manager) retryCapStart(ctx context.Context, p runtime.ProxyProvider, id int64, slug, ownIP string) bool {
+	mu := m.copyLock(id, slug)
+	mu.Lock()
+	defer mu.Unlock()
+	// The proxy as stored now: an edit while the retry waited reaches the copy, and a
+	// cap that came back stops the retry.
+	d, s, err := m.loadProxy(id)
+	if err != nil || s.CappedAt != "" {
+		return false
+	}
+	records, err := m.store.ListProxiedCopies(id)
+	if err != nil {
+		return false
+	}
+	marked := false
+	for _, r := range records {
+		marked = marked || (r.Slug == slug && r.StoppedByCap)
+	}
+	if !marked {
+		return false
+	}
+	if _, err := p.StartProxied(ctx, slug, id, m.stopTimeout(slug), specOf(d), ownIP); err != nil {
+		m.store.RecordEvent(slug, "proxied_start_error", maskSecrets(err.Error(), d))
+		return false
+	}
+	// Cleared inside the copy's lock, which stopCapped also takes, and only while the
+	// cap is still lifted: a cap that came back during the start keeps the mark, so
+	// the next stop and reset still treat the copy as the cap's.
+	m.proxy.proxyMu.Lock()
+	_, now, err := m.loadProxy(id)
+	m.proxy.proxyMu.Unlock()
+	if err == nil && now.CappedAt == "" {
+		if err := m.store.SetProxiedCopyCapped(id, slug, false); err != nil {
+			// The next pass sees it running and marked, and clears it then.
+			m.store.RecordEvent(slug, "proxied_mark_error", err.Error())
+		}
+	}
+	return true
+}
+
 func (m *Manager) stopCapped(ctx context.Context, p runtime.ProxyProvider, id int64, d proxies.Definition, targets map[string]bool) {
 	slugs := make([]string, 0, len(targets))
 	for slug := range targets {
 		slugs = append(slugs, slug)
 	}
 	sort.Strings(slugs)
 	for _, slug := range slugs {
+		m.stopCappedCopy(ctx, p, id, d, slug)
+	}
+}
+
+// stopCappedCopy stops one copy at the cap under that copy's lock, so a retry's
+// start and mark clear never interleave with the stop and its mark.
+func (m *Manager) stopCappedCopy(ctx context.Context, p runtime.ProxyProvider, id int64, d proxies.Definition, slug string) {
+	mu := m.copyLock(id, slug)
+	mu.Lock()
+	defer mu.Unlock()
+	{
 		wasRunning, err := p.StopProxied(ctx, slug, id, m.stopTimeout(slug))
 		m.proxy.lastMu.Lock()
 		if m.proxy.unstopped == nil {
@@ -760,7 +876,7 @@ func (m *Manager) stopCapped(ctx context.Context, p runtime.ProxyProvider, id in
 		}
 		m.proxy.lastMu.Unlock()
 		if err != nil || !wasRunning {
-			continue
+			return
 		}
 		// A reset or a cap change may have lifted the cap while the stop ran; then
 		// the copy goes back up instead of being marked.

internal/services/proxies_test.go

@@ -28,6 +28,10 @@ type fakeProxyProvider struct {
 	specs    []runtime.ProxySpec
 	removed  []string
 	allGone  []string
+	// startFails makes that many StartProxied calls fail before they succeed again.
+	startFails int
+	// earnerDown reports that copy's earner stopped while its tunnel runs.
+	earnerDown map[string]bool
 }
 
 func newFakeProxyProvider() *fakeProxyProvider {
@@ -52,6 +56,10 @@ func (f *fakeProxyProvider) StopProxied(_ context.Context, slug string, id int64
 }
 func (f *fakeProxyProvider) StartProxied(_ context.Context, slug string, id int64, _ int, spec runtime.ProxySpec, _ string) (runtime.Exit, error) {
 	f.starts = append(f.starts, key(slug, id))
+	if f.startFails > 0 {
+		f.startFails--
+		return runtime.Exit{}, errors.New("daemon busy")
+	}
 	f.specs = append(f.specs, spec)
 	f.running[key(slug, id)] = true
 	return runtime.Exit{IP: "203.0.113.7"}, nil
@@ -75,7 +83,11 @@ func (f *fakeProxyProvider) ProxiedStatus(context.Context) ([]runtime.ProxiedCop
 			status = "running"
 		}
 		rx, tx := f.rx[k], uint64(0)
-		out = append(out, runtime.ProxiedCopy{Slug: slug, ProxyID: id, Status: status, TunnelStatus: status, TunnelContainerID: "t-" + k, TunnelStartedAt: "s", NetRxBytes: &rx, NetTxBytes: &tx, ExitIP: "203.0.113.7"})
+		earner := status
+		if f.earnerDown[k] {
+			earner = "exited"
+		}
+		out = append(out, runtime.ProxiedCopy{Slug: slug, ProxyID: id, Status: earner, TunnelStatus: status, TunnelContainerID: "t-" + k, TunnelStartedAt: "s", NetRxBytes: &rx, NetTxBytes: &tx, ExitIP: "203.0.113.7"})
 	}
 	return out, nil
 }
@@ -415,6 +427,101 @@ func TestANewProxyWithAResetDayStartsItsPeriod(t *testing.T) {
 	}
 }
 
+// A reset whose start failed left the copy marked as the cap's and stopped until the
+// next reset, a month later. Every pass now tries again while the proxy is uncapped.
+func TestACopyAResetCouldNotStartIsStartedOnTheNextPass(t *testing.T) {
+	m, p, st := newProxyManager(t)
+	ctx := context.Background()
+	deployExample(t, st)
+	view, _ := m.SaveProxy(ctx, 0, map[string]string{"name": "Gate", "kind": "socks5", "host": "203.0.113.50", "port": "7000", "cap_gb": "1"})
+	if _, err := m.DeployProxied(ctx, "example", view.ID); err != nil {
+		t.Fatal(err)
+	}
+	k := key("example", view.ID)
+	p.rx[k] = 950_000_000
+	if _, err := m.ProxyTick(ctx); err != nil || p.running[k] {
+		t.Fatalf("capped: %v", err)
+	}
+	p.startFails = 1
+	if started, _ := m.ResetProxy(ctx, view.ID); len(started) != 0 || p.running[k] {
+		t.Fatalf("the reset's start was meant to fail: %v", started)
+	}
+	if _, err := m.ProxyTick(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if !p.running[k] {
+		t.Fatal("the next pass must start the copy the reset could not")
+	}
+	records, _ := st.ListProxiedCopies(view.ID)
+	if len(records) != 1 || records[0].StoppedByCap {
+		t.Fatalf("the mark is cleared once it runs: %+v", records)
+	}
+	// A running copy left marked (its clear failed) has the mark cleared on a pass.
+	if err := st.SetProxiedCopyCapped(view.ID, "example", true); err != nil {
+		t.Fatal(err)
+	}
+	p.starts = nil
+	if _, err := m.ProxyTick(ctx); err != nil || len(p.starts) != 0 {
+		t.Fatalf("a running marked copy must not be started: %v %v", p.starts, err)
+	}
+	if records, _ := st.ListProxiedCopies(view.ID); records[0].StoppedByCap {
+		t.Fatal("the stale mark on a running copy is cleared")
+	}
+	// A tunnel left up by a start that timed out, with the earner down, keeps the mark
+	// and is started again.
+	if err := st.SetProxiedCopyCapped(view.ID, "example", true); err != nil {
+		t.Fatal(err)
+	}
+	p.earnerDown = map[string]bool{k: true}
+	p.starts = nil
+	if _, err := m.ProxyTick(ctx); err != nil || len(p.starts) != 1 {
+		t.Fatalf("a copy with only its tunnel up must be started again: %v %v", p.starts, err)
+	}
+	p.earnerDown = nil
+	// A copy the owner stops afterwards stays stopped on the next pass.
+	if err := m.StopProxied(ctx, "example", view.ID); err != nil {
+		t.Fatal(err)
+	}
+	p.starts = nil
+	if _, err := m.ProxyTick(ctx); err != nil || len(p.starts) != 0 {
+		t.Fatalf("a copy the owner stopped was started: %v %v", p.starts, err)
+	}
+}
+
+// A retry that picked a copy before the owner stopped it reads the mark again under
+// the lock the owner's Stop holds, so the stop that returned stands.
+func TestARetryNeverUndoesAnOwnerStop(t *testing.T) {
+	m, p, st := newProxyManager(t)
+	ctx := context.Background()
+	deployExample(t, st)
+	view, _ := m.SaveProxy(ctx, 0, map[string]string{"name": "Gate", "kind": "socks5", "host": "203.0.113.50", "port": "7000"})
+	if _, err := m.DeployProxied(ctx, "example", view.ID); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.SetProxiedCopyCapped(view.ID, "example", true); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.StopProxied(ctx, "example", view.ID); err != nil {
+		t.Fatal(err)
+	}
+	p.starts = nil
+	if m.retryCapStart(ctx, p, view.ID, "example", "192.0.2.1") || len(p.starts) != 0 {
+		t.Fatalf("the retry started a copy the owner stopped: %v", p.starts)
+	}
+	// A cap that came back while the retry waited stops it too.
+	if err := st.SetProxiedCopyCapped(view.ID, "example", true); err != nil {
+		t.Fatal(err)
+	}
+	m.proxy.proxyMu.Lock()
+	_, s, _ := m.loadProxy(view.ID)
+	s.CappedAt, s.CapReason = "2026-11-07T00:00:00Z", "Gate reached its cap"
+	_ = m.saveState(view.ID, s)
+	m.proxy.proxyMu.Unlock()
+	if m.retryCapStart(ctx, p, view.ID, "example", "192.0.2.1") || len(p.starts) != 0 {
+		t.Fatalf("the retry started a copy while the cap holds: %v", p.starts)
+	}
+}
+
 // A delete the critical-data check refuses must not cost the proxied copies: the
 // refusal comes before they are removed.
 func TestARefusedDeleteKeepsTheProxiedCopies(t *testing.T) {