diff --git a/src/FSLibrary.Tests/Tests.fs b/src/FSLibrary.Tests/Tests.fs index de20cf6f..34aa0d08 100644 --- a/src/FSLibrary.Tests/Tests.fs +++ b/src/FSLibrary.Tests/Tests.fs @@ -161,6 +161,7 @@ let ctx : MissionContext = let netdata = __SOURCE_DIRECTORY__ + "/../../../data/public-network-data-2026-06-03-trimmed-located.json" + let pubkeys = __SOURCE_DIRECTORY__ + "/../../../data/tier1keys.json" let pubnetctx = { ctx with pubnetData = Some netdata; tier1Keys = Some pubkeys } @@ -704,3 +705,74 @@ type Tests(output: ITestOutputHelper) = [] member __.``QuorumIntersectionChecker mission is registered``() = Assert.True(StellarMission.allMissions.ContainsKey "QuorumIntersectionChecker") + +// A stand-in for the apiserver: rejects the first `failures` requests with 429, +// then succeeds, and counts how many times it was actually called. +type private ThrottlingStub(failures: int) = + inherit System.Net.Http.HttpMessageHandler() + let mutable calls = 0 + member __.Calls = calls + + override __.SendAsync(_req, _ct) = + calls <- calls + 1 + + let code = + if calls <= failures then + System.Net.HttpStatusCode.TooManyRequests + else + System.Net.HttpStatusCode.OK + + System.Threading.Tasks.Task.FromResult(new System.Net.Http.HttpResponseMessage(code)) + +let private sendThrough + (handler: ApiRateLimit.ThrottleRetryHandler) + (stub: ThrottlingStub) + (verb: System.Net.Http.HttpMethod) + = + handler.InnerHandler <- stub + use invoker = new System.Net.Http.HttpMessageInvoker(handler) + + let req = + new System.Net.Http.HttpRequestMessage(verb, "http://apiserver.invalid/api/v1/nodes") + + invoker.SendAsync(req, System.Threading.CancellationToken.None).Result + +[] +let ``Throttle retry rides out 429s and returns the eventual success`` () = + let stub = new ThrottlingStub(3) + let handler = new ApiRateLimit.ThrottleRetryHandler(System.TimeSpan.FromSeconds 30.0) + let resp = sendThrough handler stub System.Net.Http.HttpMethod.Get + Assert.Equal(System.Net.HttpStatusCode.OK, resp.StatusCode) + // Three rejections plus the attempt that succeeded. + Assert.Equal(4, stub.Calls) + +[] +let ``Throttle retry leaves DELETE alone so teardown stays bounded`` () = + let stub = new ThrottlingStub(5) + let handler = new ApiRateLimit.ThrottleRetryHandler(System.TimeSpan.FromSeconds 30.0) + let resp = sendThrough handler stub System.Net.Http.HttpMethod.Delete + Assert.Equal(System.Net.HttpStatusCode.TooManyRequests, resp.StatusCode) + Assert.Equal(1, stub.Calls) + +[] +let ``Throttle retry gives up at the deadline and surfaces the 429`` () = + let stub = new ThrottlingStub(1000) + let handler = new ApiRateLimit.ThrottleRetryHandler(System.TimeSpan.Zero) + let resp = sendThrough handler stub System.Net.Http.HttpMethod.Get + // The 429 must reach the caller rather than being swallowed or masked. + Assert.Equal(System.Net.HttpStatusCode.TooManyRequests, resp.StatusCode) + Assert.Equal(1, stub.Calls) + +[] +let ``Throttle retry never starts an attempt the budget cannot pay for`` () = + let stub = new ThrottlingStub(1000) + // 750ms budget: the 500ms backoff fits, the 1000ms one does not, so it stops. + let handler = new ApiRateLimit.ThrottleRetryHandler(System.TimeSpan.FromMilliseconds 750.0) + let sw = System.Diagnostics.Stopwatch.StartNew() + let resp = sendThrough handler stub System.Net.Http.HttpMethod.Get + sw.Stop() + Assert.Equal(System.Net.HttpStatusCode.TooManyRequests, resp.StatusCode) + // One wait of 500ms and two attempts; the second wait would have overrun. + Assert.Equal(2, stub.Calls) + // Stopping early is the point: it must not have slept out the full budget. + Assert.True(sw.Elapsed < System.TimeSpan.FromMilliseconds 750.0, sprintf "took %O" sw.Elapsed) diff --git a/src/FSLibrary/ApiRateLimit.fs b/src/FSLibrary/ApiRateLimit.fs index f210db61..14bb8983 100644 --- a/src/FSLibrary/ApiRateLimit.fs +++ b/src/FSLibrary/ApiRateLimit.fs @@ -5,6 +5,10 @@ module ApiRateLimit open Logging +open System.Net +open System.Net.Http +open System.Threading +open System.Threading.Tasks let mutable apiCallStopwatch = System.Diagnostics.Stopwatch.StartNew() let mutable lastApiCallTimeInMs : int64 = int64 (0) @@ -29,3 +33,55 @@ let sleepUntilNextRateLimitedApiCallTime (callsPerSec: int) = System.Threading.Thread.Sleep(toSleep) lastApiCallTimeInMs <- apiCallStopwatch.ElapsedMilliseconds + +// Retries apiserver 429s so a single rejection cannot end a mission. +type ThrottleRetryHandler(deadline: System.TimeSpan) = + inherit DelegatingHandler() + + // F# cannot call `base` from inside a task expression, so the base send needs its own member. + member private this.Send(req: HttpRequestMessage, ct: CancellationToken) = base.SendAsync(req, ct) + + override this.SendAsync(req: HttpRequestMessage, ct: CancellationToken) : Task = + // Deletes are never retried, because every delete site in this library already + // swallows failure, so retrying buys fewer orphans at the price of multiplying a + // teardown that removes hundreds of objects in sequence. + if req.Method = HttpMethod.Delete then + this.Send(req, ct) + else + let sw = System.Diagnostics.Stopwatch.StartNew() + + // Only 429 is retried, because it alone proves the request was rejected unapplied and is safe to re-send. + let rec attempt backoffMs = + task { + let! r = this.Send(req, ct) + + // Retry-After is a floor, not a replacement, or a server repeating `Retry-After: 1` pins us at one attempt per second. + let hint = + match r.Headers.RetryAfter with + | ra when not (isNull ra) && ra.Delta.HasValue -> int ra.Delta.Value.TotalMilliseconds + | _ -> 0 + + let waitMs = max backoffMs hint + + // The wait counts against the budget, so an attempt the budget cannot pay for is never started: a long Retry-After would otherwise begin one past the deadline and past HttpClientTimeout, replacing the 429 with a TaskCanceledException. + if r.StatusCode <> HttpStatusCode.TooManyRequests + || sw.Elapsed + System.TimeSpan.FromMilliseconds(float waitMs) >= deadline then + return r + else + LogWarn + "apiserver throttled %s %s (%O elapsed); retrying in %d ms" + req.Method.Method + req.RequestUri.PathAndQuery + sw.Elapsed + waitMs + + r.Dispose() + do! Task.Delay(waitMs, ct) + return! attempt (min (backoffMs * 2) 15000) + } + + task { + // A request can only be sent once unless its body is buffered first. + if not (isNull req.Content) then do! req.Content.LoadIntoBufferAsync() + return! attempt 500 + } diff --git a/src/FSLibrary/StellarSupercluster.fs b/src/FSLibrary/StellarSupercluster.fs index 2c7280fd..f042bd34 100644 --- a/src/FSLibrary/StellarSupercluster.fs +++ b/src/FSLibrary/StellarSupercluster.fs @@ -191,7 +191,12 @@ let ConnectToCluster (cfgFile: string) (nsOpt: string option) : (Kubernetes * st let clientConfig = KubernetesClientConfiguration.BuildConfigFromConfigObject(kCfg) // Disable HTTP2 to avoid intermittent issues with the cluster clientConfig.DisableHttp2 <- true - let kube = new k8s.Kubernetes(clientConfig) + // Rides out apiserver 429s for every call this client makes, and must stay + // well under clientConfig.HttpClientTimeout (100s), which bounds the whole + // handler chain and surfaces as a TaskCanceledException that loses the 429. + let kube = + new k8s.Kubernetes(clientConfig, new ApiRateLimit.ThrottleRetryHandler(System.TimeSpan.FromSeconds 60.0)) + (kube, ns) // Prints the stellar-core StatefulSets and Pods on the provided cluster