diff --git a/CHANGELOG.md b/CHANGELOG.md index 38d1c5e..359240e 100644 --- a/CHANGELOG.md +++ b/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/. diff --git a/docs/proxies.md b/docs/proxies.md index 625f4ca..9dcb518 100644 --- a/docs/proxies.md +++ b/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 diff --git a/internal/services/proxies.go b/internal/services/proxies.go index e3d358b..047cbca 100644 --- a/internal/services/proxies.go +++ b/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,6 +757,90 @@ 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 { @@ -743,6 +848,17 @@ func (m *Manager) stopCapped(ctx context.Context, p runtime.ProxyProvider, id in } 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. diff --git a/internal/services/proxies_test.go b/internal/services/proxies_test.go index df073e5..00f6200 100644 --- a/internal/services/proxies_test.go +++ b/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) {