penguinehis/DragonCoreSSH-NewWEB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix DNSTT Default

Browse Source
This commit is contained in:
penguinehis
2026-07-24 14:46:42 -03:00
parent 7c51ea3f86
commit 1576ce9038
11 changed files with 528 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+4 -4
View File
@@ -199,7 +199,7 @@ Também é possível editar diretamente o `config.json`:
```json
"dnstt": {
"domain": "t.example.com",
"udp_listen": "[::]:5300",
"udp_listen": "0.0.0.0:5300",
"privkey_file": "/opt/sshpanel/dnstt.key",
"auto_restart_interval": "6h",
"auto_restart_grace": "2s"
@@ -219,7 +219,7 @@ Exemplo:
"t.example.com",
"t.local.lan"
],
"udp_listen": "[::]:5300",
"udp_listen": "0.0.0.0:5300",
"privkey_file": "/opt/sshpanel/dnstt.key"
}
```
@@ -738,7 +738,7 @@ You can also edit `config.json` directly:
```json
"dnstt": {
"domain": "t.example.com",
"udp_listen": "[::]:5300",
"udp_listen": "0.0.0.0:5300",
"privkey_file": "/opt/sshpanel/dnstt.key",
"auto_restart_interval": "6h",
"auto_restart_grace": "2s"
@@ -758,7 +758,7 @@ Example:
"t.example.com",
"t.local.lan"
],
"udp_listen": "[::]:5300",
"udp_listen": "0.0.0.0:5300",
"privkey_file": "/opt/sshpanel/dnstt.key"
}
```
admin/assets/js/06-servers.js
+2 -2
View File
@@ -426,7 +426,7 @@ async function loadManagedServerConfig(id) {
toggleManagedDnsttFields(hasDnstt);
const d = c.dnstt || {};
document.getElementById("managedCfgDnsttDomains").value = dnsttDomainsText(d);
document.getElementById("managedCfgDnsttUDP").value = d.udp_listen || "";
document.getElementById("managedCfgDnsttUDP").value = d.udp_listen || "0.0.0.0:5300";
document.getElementById("managedCfgDnsttFakeEnabled").checked = !!d.fake_dns_enabled;
document.getElementById("managedCfgDnsttFakeListen").value = d.fake_dns_listen || "";
document.getElementById("managedCfgDnsttFakeDomain").value = d.fake_dns_domain || "t.local.lan";
@@ -494,7 +494,7 @@ function managedConfigFromForm() {
dnstt: document.getElementById("managedCfgDnsttEnabled").checked ? {
domain: dnsttDomains[0] || "",
domains: dnsttDomains,
udp_listen: document.getElementById("managedCfgDnsttUDP").value.trim(),
udp_listen: document.getElementById("managedCfgDnsttUDP").value.trim() || "0.0.0.0:5300",
fake_dns_enabled: document.getElementById("managedCfgDnsttFakeEnabled").checked,
fake_dns_listen: document.getElementById("managedCfgDnsttFakeListen").value.trim(),
fake_dns_domain: document.getElementById("managedCfgDnsttFakeDomain").value.trim(),
admin/assets/js/08-server-config.js
+2 -2
View File
@@ -122,7 +122,7 @@ async function loadServerConfig() {
toggleDnsttFields(hasDnstt);
const d = c.dnstt || {};
document.getElementById("cfgDnsttDomains").value = dnsttDomainsText(d);
document.getElementById("cfgDnsttUDP").value = d.udp_listen || "";
document.getElementById("cfgDnsttUDP").value = d.udp_listen || "0.0.0.0:5300";
document.getElementById("cfgDnsttFakeEnabled").checked = !!d.fake_dns_enabled;
document.getElementById("cfgDnsttFakeListen").value = d.fake_dns_listen || "";
document.getElementById("cfgDnsttFakeDomain").value = d.fake_dns_domain || "t.local.lan";
@@ -201,7 +201,7 @@ async function saveServerConfig() {
dnstt: document.getElementById("cfgDnsttEnabled").checked ? {
domain: dnsttDomains[0] || "",
domains: dnsttDomains,
udp_listen: document.getElementById("cfgDnsttUDP").value.trim(),
udp_listen: document.getElementById("cfgDnsttUDP").value.trim() || "0.0.0.0:5300",
fake_dns_enabled: document.getElementById("cfgDnsttFakeEnabled").checked,
fake_dns_listen: document.getElementById("cfgDnsttFakeListen").value.trim(),
fake_dns_domain: document.getElementById("cfgDnsttFakeDomain").value.trim(),
admin/index.html
+2 -2
View File
@@ -867,7 +867,7 @@
</div>
<div id="managedDnsttFields" class="form-grid" style="opacity:.4;pointer-events:none;">
<div class="field" style="grid-column:1/-1"><label>NS / Root Domains <span class="hint">one per line</span></label><textarea id="managedCfgDnsttDomains" rows="3" placeholder="t.example.com&#10;t.local.lan"></textarea></div>
<div class="field"><label>UDP Listen</label><input type="text" id="managedCfgDnsttUDP" placeholder="[::]:5300"/></div>
<div class="field"><label>UDP Listen</label><input type="text" id="managedCfgDnsttUDP" placeholder="0.0.0.0:5300"/></div>
<label style="font-size:.73rem;display:flex;align-items:center;gap:5px;cursor:pointer;grid-column:1/-1"><input type="checkbox" id="managedCfgDnsttFakeEnabled"/> Built-in Local DNS / Fake DNS</label>
<div class="field"><label>Local DNS Listen <span class="hint">IPv6 ok</span></label><input type="text" id="managedCfgDnsttFakeListen" placeholder="[2001:db8::1234]:53"/></div>
<div class="field"><label>Local DNS Domain</label><input type="text" id="managedCfgDnsttFakeDomain" placeholder="t.local.lan"/></div>
@@ -1344,7 +1344,7 @@
</div>
<div class="field">
<label>UDP Listen</label>
<input type="text" id="cfgDnsttUDP" placeholder="[::]:5300"/>
<input type="text" id="cfgDnsttUDP" placeholder="0.0.0.0:5300"/>
</div>
<label style="font-size:.73rem;display:flex;align-items:center;gap:5px;cursor:pointer;grid-column:1/-1">
<input type="checkbox" id="cfgDnsttFakeEnabled"/> Built-in Local DNS / Fake DNS
config_safety.go
+28 -4
View File
@@ -11,7 +11,7 @@ import (
const (
defaultMainListen = "0.0.0.0:80"
defaultExtraListen = "0.0.0.0:8080"
defaultDNSTTListen = "[::]:5300"
defaultDNSTTListen = "0.0.0.0:5300"
defaultUDPGWListen = "0.0.0.0:7400"
)
@@ -109,9 +109,10 @@ func normalizeRuntimePorts(cfg *Config) []string {
}
}
cfg.DNSTT.UDPListen = strings.TrimSpace(cfg.DNSTT.UDPListen)
if cfg.DNSTT.UDPListen == "" {
cfg.DNSTT.UDPListen = defaultDNSTTListen
var migratedLegacyDNSTTWildcard bool
cfg.DNSTT.UDPListen, migratedLegacyDNSTTWildcard = normalizeDNSTTListenDefault(cfg.DNSTT.UDPListen)
if migratedLegacyDNSTTWildcard {
warn("DNSTT legacy default [::]:5300 is IPv6-only; using IPv4 default %s", cfg.DNSTT.UDPListen)
}
if err := udpAddrAvailableForDNSTT(cfg.DNSTT.UDPListen); err != nil {
old := cfg.DNSTT.UDPListen
@@ -295,6 +296,29 @@ func normalizeDNSTTDomainList(primary string, domains []string) []string {
return out
}
// normalizeDNSTTListenDefault keeps explicit IPv4 and concrete IPv6 listeners,
// but migrates the old wildcard IPv6 default. listenDNSTTPacket deliberately
// opens IPv6 addresses with udp6, so [::]:5300 never receives IPv4 queries.
// Existing installations commonly inherited that value from the old default;
// moving only that wildcard/default-port combination makes them work after an
// update without changing intentionally selected IPv6 interface addresses.
func normalizeDNSTTListenDefault(addr string) (string, bool) {
addr = strings.TrimSpace(addr)
if addr == "" {
return defaultDNSTTListen, false
}
host, port, err := net.SplitHostPort(addr)
if err != nil || port != "5300" {
return addr, false
}
ip := net.ParseIP(strings.Trim(host, "[]"))
if ip != nil && ip.To4() == nil && ip.IsUnspecified() {
return defaultDNSTTListen, true
}
return addr, false
}
func udpAddrAvailableForDNSTT(addr string) error {
if addr == "" {
return nil
config_safety_test.go
+30
View File
@@ -0,0 +1,30 @@
package main
import "testing"
func TestNormalizeDNSTTListenDefault(t *testing.T) {
tests := []struct {
name string
input string
want string
migrated bool
}{
{name: "empty uses IPv4 default", input: "", want: "0.0.0.0:5300"},
{name: "whitespace uses IPv4 default", input: " ", want: "0.0.0.0:5300"},
{name: "legacy IPv6 wildcard migrates", input: "[::]:5300", want: "0.0.0.0:5300", migrated: true},
{name: "expanded legacy wildcard migrates", input: "[0:0:0:0:0:0:0:0]:5300", want: "0.0.0.0:5300", migrated: true},
{name: "explicit IPv4 remains", input: "192.0.2.10:53", want: "192.0.2.10:53"},
{name: "IPv4 wildcard remains", input: "0.0.0.0:5300", want: "0.0.0.0:5300"},
{name: "concrete IPv6 remains", input: "[2001:db8::10]:53", want: "[2001:db8::10]:53"},
{name: "IPv6 wildcard on custom port remains", input: "[::]:5301", want: "[::]:5301"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, migrated := normalizeDNSTTListenDefault(tt.input)
if got != tt.want || migrated != tt.migrated {
t.Fatalf("normalizeDNSTTListenDefault(%q) = (%q, %v), want (%q, %v)", tt.input, got, migrated, tt.want, tt.migrated)
}
})
}
}
dnstt_integration.go
+1 -4
View File
@@ -595,11 +595,8 @@ func startDNSTTInstance(cfg *DNSTTConfig, sshConf *ssh.ServerConfig) error {
return err
}
udpListen := cfg.UDPListen
if udpListen == "" {
udpListen = defaultDNSTTListen
udpListen, _ := normalizeDNSTTListenDefault(cfg.UDPListen)
cfg.UDPListen = udpListen
}
fakeDomains := domains
if cfg.FakeDNSEnabled {
quota_hardening_test.go
+196 -6
View File
@@ -7,6 +7,7 @@ import (
"net"
"net/http"
"net/http/httptest"
"os"
"strings"
"sync"
"sync/atomic"
@@ -240,15 +241,21 @@ func TestXHTTPSessionsIgnoreLegacyGlobalCapAndReleaseCounters(t *testing.T) {
}
}
func TestNegativeXHTTPSessionLimitMeansUnlimited(t *testing.T) {
if got := (&nativeInbound{}).xhttpMaxActiveSessions(); got != 0 {
t.Fatalf("unlimited XHTTP session limit normalized to %d", got)
func TestXHTTPSessionSafetyWindowIsAboveProductionScale(t *testing.T) {
if got := (&nativeInbound{}).xhttpMaxActiveSessions(); got != fixedNativeMaxXHTTPSessions {
t.Fatalf("XHTTP session safety window = %d, want %d", got, fixedNativeMaxXHTTPSessions)
}
if got := nativeXHTTPSessionLimit(); got < 8_000 {
t.Fatalf("XHTTP session safety window = %d, want room for at least 8K users", got)
}
}
func TestNativeHTTP2StreamsIgnoreLegacyRequestCeiling(t *testing.T) {
if got := nativeHTTP2MaxConcurrentStreams(); got != ^uint32(0) {
t.Fatalf("HTTP/2 stream setting = %d, want unlimited uint32 range", got)
func TestNativeHTTP2StreamsUseFiniteTransportBackpressure(t *testing.T) {
if got := nativeHTTP2MaxConcurrentStreams(); got != fixedNativeHTTP2ConcurrentStreams {
t.Fatalf("HTTP/2 stream setting = %d, want %d", got, fixedNativeHTTP2ConcurrentStreams)
}
if got := nativeHTTP2MaxConcurrentStreams(); got < 1024 {
t.Fatalf("HTTP/2 stream setting = %d, too small for XHTTP packet bursts", got)
}
if got := nativeMuxMaxSessionLimit(); got != 64 {
t.Fatalf("per-transport Mux session guard = %d, want 64", got)
@@ -265,6 +272,20 @@ func TestXHTTPHandlerDoesNotApplyWebRequestCeiling(t *testing.T) {
}
}
func TestXHTTPHandlerSafetySlotIsReleased(t *testing.T) {
before := nativeXHTTPRequests.Load()
ib := &nativeInbound{transport: "xhttp", path: "/"}
req := httptest.NewRequest(http.MethodGet, "/", nil)
rec := httptest.NewRecorder()
ib.ServeHTTP(rec, req)
if rec.Code != http.StatusBadRequest {
t.Fatalf("empty XHTTP request status = %d, want 400", rec.Code)
}
if got := nativeXHTTPRequests.Load(); got != before {
t.Fatalf("XHTTP handler counter after return = %d, want %d", got, before)
}
}
func TestPersistedXHTTPAdmissionTuningIsAlwaysUnlimited(t *testing.T) {
got := normalizeNativeXrayTuning(&XrayNativeTuning{
MuxGlobalSessions: 8192,
@@ -411,6 +432,80 @@ func TestXHTTPUploadQueueBackpressuresInsteadOfRejectingBurst(t *testing.T) {
}
}
func TestXHTTPGlobalMemoryBackpressureWakesWaitersFIFO(t *testing.T) {
before := nativeXHTTPBufferedBytes.Load()
fillBytes := nativeXHTTPMaxBufferedGlobalBytes - before
filler, ok := acquireNativeXHTTPMemory(fillBytes)
if !ok {
t.Fatal("failed to fill XHTTP memory budget for waiter test")
}
defer filler.release()
type result struct {
lease *nativeXHTTPMemoryLease
err error
}
startWaiter := func() <-chan result {
done := make(chan result, 1)
go func() {
lease, err := acquireNativeXHTTPMemoryContext(context.Background(), nativeXHTTPMinPacketAccountingBytes)
done <- result{lease: lease, err: err}
}()
return done
}
waitForWaiters := func(want int) {
t.Helper()
deadline := time.Now().Add(time.Second)
for {
nativeXHTTPMemoryWait.Lock()
got := nativeXHTTPMemoryWait.queued
nativeXHTTPMemoryWait.Unlock()
if got == want {
return
}
if time.Now().After(deadline) {
t.Fatalf("memory waiters = %d, want %d", got, want)
}
time.Sleep(time.Millisecond)
}
}
firstDone := startWaiter()
waitForWaiters(1)
secondDone := startWaiter()
waitForWaiters(2)
filler.shrink(fillBytes - nativeXHTTPMinPacketAccountingBytes)
first := <-firstDone
if first.err != nil || first.lease == nil {
t.Fatalf("first memory waiter = (%v, %v)", first.lease, first.err)
}
select {
case second := <-secondDone:
if second.lease != nil {
second.lease.release()
}
t.Fatalf("second waiter woke before FIFO capacity was released: %v", second.err)
case <-time.After(25 * time.Millisecond):
}
first.lease.release()
select {
case second := <-secondDone:
if second.err != nil || second.lease == nil {
t.Fatalf("second memory waiter = (%v, %v)", second.lease, second.err)
}
second.lease.release()
case <-time.After(time.Second):
t.Fatal("second memory waiter did not wake after first released")
}
filler.release()
if got := nativeXHTTPBufferedBytes.Load(); got != before {
t.Fatalf("FIFO waiter test leaked %d buffered bytes (baseline %d)", got, before)
}
}
func TestXHTTPReassemblyHasNoPacketRequestCountCeiling(t *testing.T) {
before := nativeXHTTPBufferedBytes.Load()
q := newNativeXHTTPUploadQueue(1, 4*nativeXHTTPMinPacketAccountingBytes)
@@ -678,3 +773,98 @@ func TestNativeXHTTPQueueCloseClosesQueuedStreamReader(t *testing.T) {
t.Fatal("queued stream reader was not closed during queue shutdown")
}
}
func TestNativeXHTTPQueueSkipsEmptyPacketsWithoutZeroProgressRead(t *testing.T) {
before := nativeXHTTPBufferedBytes.Load()
q := newNativeXHTTPUploadQueue(2, 2*nativeXHTTPMinPacketAccountingBytes)
defer q.close()
for seq, payload := range [][]byte{nil, []byte("x")} {
accounted := nativeXHTTPAccountedPacketBytes(int64(len(payload)))
lease, ok := acquireNativeXHTTPMemory(accounted)
if !ok {
t.Fatal("failed to reserve packet memory")
}
if err := q.push(context.Background(), nativeXHTTPPacket{Payload: payload, Seq: uint64(seq)}, lease); err != nil {
lease.release()
t.Fatalf("queue packet %d: %v", seq, err)
}
lease.release()
}
buf := make([]byte, 1)
n, err := q.Read(buf)
if err != nil || n != 1 || string(buf[:n]) != "x" {
t.Fatalf("queue read after empty packet = (%d, %v, %q), want (1, nil, x)", n, err, buf[:n])
}
q.close()
if got := nativeXHTTPBufferedBytes.Load(); got != before {
t.Fatalf("empty-packet test leaked %d buffered bytes (baseline %d)", got, before)
}
}
type deadlineUnblockingResponseWriter struct {
header http.Header
writeStart chan struct{}
unblock chan struct{}
startOnce sync.Once
unblockOnce sync.Once
}
func newDeadlineUnblockingResponseWriter() *deadlineUnblockingResponseWriter {
return &deadlineUnblockingResponseWriter{
header: make(http.Header),
writeStart: make(chan struct{}),
unblock: make(chan struct{}),
}
}
func (w *deadlineUnblockingResponseWriter) Header() http.Header { return w.header }
func (w *deadlineUnblockingResponseWriter) WriteHeader(int) {}
func (w *deadlineUnblockingResponseWriter) Flush() {}
func (w *deadlineUnblockingResponseWriter) Write([]byte) (int, error) {
w.startOnce.Do(func() { close(w.writeStart) })
<-w.unblock
return 0, os.ErrDeadlineExceeded
}
func (w *deadlineUnblockingResponseWriter) SetWriteDeadline(deadline time.Time) error {
if !deadline.IsZero() && !deadline.After(time.Now().Add(10*time.Millisecond)) {
w.unblockOnce.Do(func() { close(w.unblock) })
}
return nil
}
func TestNativeXHTTPResponseCloseInterruptsStalledWrite(t *testing.T) {
underlying := newDeadlineUnblockingResponseWriter()
writer := newNativeXHTTPResponseWriter(underlying)
writeDone := make(chan error, 1)
go func() {
_, err := writer.Write([]byte("blocked"))
writeDone <- err
}()
select {
case <-underlying.writeStart:
case <-time.After(time.Second):
t.Fatal("response write did not start")
}
closeDone := make(chan struct{})
go func() {
writer.close()
close(closeDone)
}()
select {
case <-closeDone:
case <-time.After(time.Second):
t.Fatal("response close blocked behind stalled write")
}
select {
case err := <-writeDone:
if !errors.Is(err, os.ErrDeadlineExceeded) {
t.Fatalf("stalled write error = %v, want deadline exceeded", err)
}
case <-time.After(time.Second):
t.Fatal("stalled response write was not interrupted")
}
}
xray_native_safety.go
+15 -8
View File
@@ -36,8 +36,10 @@ func init() {
var (
nativeTransportConnections atomic.Int64
nativeXHTTPSessions atomic.Int64
nativeXHTTPRequests atomic.Int64
nativeClientConnsRejected atomic.Int64
nativePreAuthRejected atomic.Int64
nativeXHTTPRejected atomic.Int64
nativeTransportAccepting atomic.Bool
nativeTransportRegistry = struct {
@@ -47,8 +49,8 @@ var (
)
// acquireNativeCounter tracks a counted resource and returns an exactly-once
// release function. Native transport/XHTTP admission calls it with limit=0
// because VPN traffic must not be rejected by a global website-style ceiling.
// release function. Limits here are simultaneous resource-safety windows, not
// traffic-volume or request-rate ceilings.
func acquireNativeCounter(active *atomic.Int64, limit int) (func(), bool) {
for {
current := active.Load()
@@ -96,11 +98,15 @@ func logNativePreAuthRejection(format string, args ...interface{}) {
}
func acquireNativeTransportConnection() (func(), bool) {
return acquireNativeCounter(&nativeTransportConnections, 0)
return acquireNativeCounter(&nativeTransportConnections, nativeTransportConnectionLimit())
}
func acquireNativeXHTTPSession() (func(), bool) {
return acquireNativeCounter(&nativeXHTTPSessions, 0)
return acquireNativeCounter(&nativeXHTTPSessions, nativeXHTTPSessionLimit())
}
func acquireNativeXHTTPRequest() (func(), bool) {
return acquireNativeCounter(&nativeXHTTPRequests, nativeXHTTPRequestLimit())
}
func configureNativeTransportSocket(c net.Conn) {
@@ -192,8 +198,9 @@ func registerTrackedNativeTransportConn(c net.Conn, release func()) (net.Conn, b
return counted, true
}
// waitWrapTrackedNativeTransportConn is used by raw native accept loops. Global
// admission is unlimited; the loop remains only to coordinate listener shutdown.
// waitWrapTrackedNativeTransportConn is used by raw native accept loops. Waiting
// here, before another connection is admitted to the protocol handler, applies
// socket/kernel backpressure instead of creating an unbounded goroutine backlog.
func waitWrapTrackedNativeTransportConn(c net.Conn) (net.Conn, bool) {
if c == nil {
return nil, false
@@ -231,8 +238,8 @@ func closeAllNativeTransportConnections() {
}
// nativeTrackingListener registers every accepted XHTTP socket so a live
// stop/reload can close it. It counts sockets for diagnostics but never rejects
// or delays one because of a global application limit.
// stop/reload can close it. It reserves capacity before Accept so overload stays
// in the kernel accept queue rather than allocating more Go handlers.
type nativeTrackingListener struct {
net.Listener
}
xray_native_tuning.go
+19 -4
View File
@@ -37,6 +37,18 @@ const (
// buffers. Operators may request more, up to the hard cap enforced there.
defaultNativeXHTTPBufferedPosts = 64
// These are simultaneous resource-safety windows, not request-rate limits.
// They are deliberately far above the expected 6-8K connected-user load, but
// finite so a reconnect storm, broken CDN, or hostile client cannot retain an
// unbounded number of sockets, HTTP handlers, sessions, and goroutine stacks.
// Transport Accept waits at capacity (kernel backpressure); XHTTP overloads
// receive 503 rather than the web-rate-limit semantics of 429.
fixedNativeMaxTransportConnections = 65536
fixedNativeMaxXHTTPRequests = 65536
fixedNativeMaxXHTTPSessions = 65536
fixedNativeHTTP2ConcurrentStreams = 4096
fixedNativeXHTTPWriteTimeoutMS = 60 * 1000
// Backstop reaper for connected XHTTP VPN sessions. The stream-down GET's
// request context is the primary lifetime owner, but behind a CDN that context
// frequently never fires when a client silently drops (mobile networks, CDN
@@ -111,10 +123,13 @@ func nativeMuxUDPReadBufferSize() int { return fixedNativeMuxUDPReadBuffer }
func nativeMuxUDPWriteBufferSize() int { return fixedNativeMuxUDPWriteBuffer }
func nativeXHTTPBufferedPostLimit() int { return defaultNativeXHTTPBufferedPosts }
func nativeHTTP2MaxConcurrentStreams() uint32 {
// x/net/http2 otherwise installs its own finite default when this is zero.
// Advertise the protocol's full uint32 range so ordinary packet-up bursts can
// never be refused by a website-style concurrent-stream setting.
return ^uint32(0)
return fixedNativeHTTP2ConcurrentStreams
}
func nativeTransportConnectionLimit() int { return fixedNativeMaxTransportConnections }
func nativeXHTTPRequestLimit() int { return fixedNativeMaxXHTTPRequests }
func nativeXHTTPSessionLimit() int { return fixedNativeMaxXHTTPSessions }
func nativeXHTTPWriteTimeout() time.Duration {
return fixedNativeXHTTPWriteTimeoutMS * time.Millisecond
}
func nativeMuxUDPIdleTimeout() time.Duration {
return fixedNativeMuxUDPIdleMS * time.Millisecond
xray_xhttp.go
+167 -40
View File
@@ -41,10 +41,18 @@ var (
errNativeXHTTPUploadBufferFull = errors.New("xhttp upload buffer limit reached")
nativeXHTTPMemoryWait = struct {
sync.Mutex
changed chan struct{}
}{changed: make(chan struct{})}
waiters []*nativeXHTTPMemoryWaiter
head int
queued int
}{}
)
type nativeXHTTPMemoryWaiter struct {
bytes int64
ready chan struct{}
granted bool
}
const (
xhttpPlacementPath = "path"
xhttpPlacementQuery = "query"
@@ -328,6 +336,15 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
releaseRequest, ok := acquireNativeXHTTPRequest()
if !ok {
logNativeLimitRejection("simultaneous XHTTP handlers", &nativeXHTTPRejected, nativeXHTTPRequestLimit())
w.Header().Set("Retry-After", "1")
http.Error(w, "xhttp transport temporarily busy", http.StatusServiceUnavailable)
return
}
defer releaseRequest()
sessionID, seqStr := ib.extractXHTTPMeta(r, base)
if len(sessionID) > nativeXHTTPMaxSessionIDBytes || len(seqStr) > nativeXHTTPMaxSequenceBytes {
logNativePreAuthRejection("native xray: xhttp reject inbound=%q reason=metadata-size remote=%s", ib.tag, r.RemoteAddr)
@@ -560,7 +577,13 @@ func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *n
s.touch()
return s
}
releaseSlot, _ := acquireNativeXHTTPSession()
releaseSlot, ok := acquireNativeXHTTPSession()
if !ok {
logNativeLimitRejection("simultaneous XHTTP sessions", &nativeXHTTPRejected, nativeXHTTPSessionLimit())
w.Header().Set("Retry-After", "1")
http.Error(w, "xhttp session capacity temporarily busy", http.StatusServiceUnavailable)
return nil
}
s := &nativeXHTTPSession{
id: id,
queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts, nativeXHTTPMaxBufferedSessionBytes),
@@ -576,7 +599,7 @@ func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *n
}
func (ib *nativeInbound) xhttpMaxActiveSessions() int {
return 0
return nativeXHTTPSessionLimit()
}
func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) {
@@ -1048,12 +1071,19 @@ func (c *nativeXHTTPConn) SetReadDeadline(t time.Time) error {
return nil
}
func (c *nativeXHTTPConn) SetWriteDeadline(time.Time) error { return nil }
func (c *nativeXHTTPConn) SetWriteDeadline(t time.Time) error {
if dw, ok := c.writer.(interface{ SetWriteDeadline(time.Time) error }); ok {
return dw.SetWriteDeadline(t)
}
return nil
}
type nativeXHTTPResponseWriter struct {
mu sync.Mutex
writeMu sync.Mutex
stateMu sync.Mutex
w http.ResponseWriter
closed bool
deadline time.Time
}
func newNativeXHTTPResponseWriter(w http.ResponseWriter) *nativeXHTTPResponseWriter {
@@ -1061,22 +1091,53 @@ func newNativeXHTTPResponseWriter(w http.ResponseWriter) *nativeXHTTPResponseWri
}
func (w *nativeXHTTPResponseWriter) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
if w.closed {
w.writeMu.Lock()
defer w.writeMu.Unlock()
w.stateMu.Lock()
closed := w.closed
deadline := w.deadline
w.stateMu.Unlock()
if closed {
return 0, io.ErrClosedPipe
}
safetyDeadline := time.Now().Add(nativeXHTTPWriteTimeout())
if deadline.IsZero() || deadline.After(safetyDeadline) {
deadline = safetyDeadline
}
controller := http.NewResponseController(w.w)
if err := controller.SetWriteDeadline(deadline); err != nil && !errors.Is(err, http.ErrNotSupported) {
return 0, err
}
n, err := w.w.Write(p)
if err == nil {
flushHTTP(w.w)
if flushErr := controller.Flush(); flushErr != nil && !errors.Is(flushErr, http.ErrNotSupported) {
err = flushErr
}
}
return n, err
}
func (w *nativeXHTTPResponseWriter) close() {
w.mu.Lock()
// Do not wait for writeMu: Close is commonly called by the request-context
// watcher specifically because a CDN write is stalled. Mark the writer closed
// and force the active net/http write deadline to expire so Write returns.
w.stateMu.Lock()
w.closed = true
w.mu.Unlock()
w.stateMu.Unlock()
_ = http.NewResponseController(w.w).SetWriteDeadline(time.Now())
}
func (w *nativeXHTTPResponseWriter) SetWriteDeadline(t time.Time) error {
w.stateMu.Lock()
w.deadline = t
w.stateMu.Unlock()
err := http.NewResponseController(w.w).SetWriteDeadline(t)
if errors.Is(err, http.ErrNotSupported) {
return nil
}
return err
}
// nativeXHTTPMemoryLease reserves from a process-wide byte budget before a
@@ -1091,17 +1152,16 @@ func acquireNativeXHTTPMemory(n int64) (*nativeXHTTPMemoryLease, bool) {
if n <= 0 {
return &nativeXHTTPMemoryLease{}, true
}
for {
nativeXHTTPMemoryWait.Lock()
defer nativeXHTTPMemoryWait.Unlock()
current := nativeXHTTPBufferedBytes.Load()
if current > nativeXHTTPMaxBufferedGlobalBytes-n {
if nativeXHTTPMemoryWait.queued != 0 || current > nativeXHTTPMaxBufferedGlobalBytes-n {
logNativeLimitRejection("XHTTP buffered upload bytes", &nativeXHTTPBufferRejected, nativeXHTTPMaxBufferedGlobalBytes)
return nil, false
}
if nativeXHTTPBufferedBytes.CompareAndSwap(current, current+n) {
nativeXHTTPBufferedBytes.Store(current + n)
return &nativeXHTTPMemoryLease{bytes: n}, true
}
}
}
// acquireNativeXHTTPMemoryContext applies process-wide memory backpressure.
// Unlike the old fail-fast admission path, a legitimate tunnel burst waits for
@@ -1113,27 +1173,42 @@ func acquireNativeXHTTPMemoryContext(ctx context.Context, n int64) (*nativeXHTTP
if n > nativeXHTTPMaxBufferedGlobalBytes {
return nil, errNativeXHTTPUploadBufferFull
}
for {
waiter := &nativeXHTTPMemoryWaiter{bytes: n, ready: make(chan struct{})}
nativeXHTTPMemoryWait.Lock()
current := nativeXHTTPBufferedBytes.Load()
if current <= nativeXHTTPMaxBufferedGlobalBytes-n && nativeXHTTPBufferedBytes.CompareAndSwap(current, current+n) {
if nativeXHTTPMemoryWait.queued == 0 && current <= nativeXHTTPMaxBufferedGlobalBytes-n {
nativeXHTTPBufferedBytes.Store(current + n)
nativeXHTTPMemoryWait.Unlock()
return &nativeXHTTPMemoryLease{bytes: n}, nil
}
nativeXHTTPMemoryWait.waiters = append(nativeXHTTPMemoryWait.waiters, waiter)
nativeXHTTPMemoryWait.queued++
nativeXHTTPMemoryWait.Unlock()
nativeXHTTPMemoryWait.Lock()
// Recheck while holding the generation lock so a release cannot happen
// between the failed check and subscribing to the notification channel.
current = nativeXHTTPBufferedBytes.Load()
if current <= nativeXHTTPMaxBufferedGlobalBytes-n {
nativeXHTTPMemoryWait.Unlock()
continue
}
changed := nativeXHTTPMemoryWait.changed
nativeXHTTPMemoryWait.Unlock()
select {
case <-changed:
case <-waiter.ready:
return &nativeXHTTPMemoryLease{bytes: n}, nil
case <-ctx.Done():
return nil, ctx.Err()
nativeXHTTPMemoryWait.Lock()
if waiter.granted {
current := nativeXHTTPBufferedBytes.Load() - n
if current < 0 {
current = 0
}
nativeXHTTPBufferedBytes.Store(current)
} else {
for i := nativeXHTTPMemoryWait.head; i < len(nativeXHTTPMemoryWait.waiters); i++ {
candidate := nativeXHTTPMemoryWait.waiters[i]
if candidate == waiter {
nativeXHTTPMemoryWait.waiters[i] = nil
nativeXHTTPMemoryWait.queued--
break
}
}
}
grantNativeXHTTPMemoryWaitersLocked()
nativeXHTTPMemoryWait.Unlock()
return nil, ctx.Err()
}
}
@@ -1141,20 +1216,69 @@ func releaseNativeXHTTPMemory(n int64) {
if n <= 0 {
return
}
for {
nativeXHTTPMemoryWait.Lock()
current := nativeXHTTPBufferedBytes.Load()
next := current - n
if next < 0 {
next = 0
}
if nativeXHTTPBufferedBytes.CompareAndSwap(current, next) {
nativeXHTTPMemoryWait.Lock()
close(nativeXHTTPMemoryWait.changed)
nativeXHTTPMemoryWait.changed = make(chan struct{})
nativeXHTTPBufferedBytes.Store(next)
grantNativeXHTTPMemoryWaitersLocked()
nativeXHTTPMemoryWait.Unlock()
}
// grantNativeXHTTPMemoryWaitersLocked wakes only the FIFO waiters whose exact
// reservations now fit. The former broadcast channel woke every blocked HTTP
// handler after every tiny release, creating a thundering herd and sustained
// multi-core CPU usage while the 128 MB budget was full.
func grantNativeXHTTPMemoryWaitersLocked() {
for nativeXHTTPMemoryWait.queued > 0 {
for nativeXHTTPMemoryWait.head < len(nativeXHTTPMemoryWait.waiters) &&
nativeXHTTPMemoryWait.waiters[nativeXHTTPMemoryWait.head] == nil {
nativeXHTTPMemoryWait.head++
}
if nativeXHTTPMemoryWait.head >= len(nativeXHTTPMemoryWait.waiters) {
nativeXHTTPMemoryWait.waiters = nil
nativeXHTTPMemoryWait.head = 0
nativeXHTTPMemoryWait.queued = 0
return
}
waiter := nativeXHTTPMemoryWait.waiters[nativeXHTTPMemoryWait.head]
current := nativeXHTTPBufferedBytes.Load()
if current > nativeXHTTPMaxBufferedGlobalBytes-waiter.bytes {
compactNativeXHTTPMemoryWaitersLocked()
return
}
nativeXHTTPMemoryWait.waiters[nativeXHTTPMemoryWait.head] = nil
nativeXHTTPMemoryWait.head++
nativeXHTTPMemoryWait.queued--
nativeXHTTPBufferedBytes.Store(current + waiter.bytes)
waiter.granted = true
close(waiter.ready)
}
compactNativeXHTTPMemoryWaitersLocked()
}
func compactNativeXHTTPMemoryWaitersLocked() {
head := nativeXHTTPMemoryWait.head
if head == 0 {
return
}
if nativeXHTTPMemoryWait.queued == 0 {
nativeXHTTPMemoryWait.waiters = nil
nativeXHTTPMemoryWait.head = 0
return
}
if head < 1024 && head*2 < len(nativeXHTTPMemoryWait.waiters) {
return
}
remaining := copy(nativeXHTTPMemoryWait.waiters, nativeXHTTPMemoryWait.waiters[head:])
for i := remaining; i < len(nativeXHTTPMemoryWait.waiters); i++ {
nativeXHTTPMemoryWait.waiters[i] = nil
}
nativeXHTTPMemoryWait.waiters = nativeXHTTPMemoryWait.waiters[:remaining]
nativeXHTTPMemoryWait.head = 0
}
func (l *nativeXHTTPMemoryLease) shrink(n int64) {
@@ -1427,6 +1551,7 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
return reader.Read(b)
}
for {
select {
case <-q.closed:
return 0, io.EOF
@@ -1458,6 +1583,11 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
packet := heap.Pop(&q.heap).(nativeXHTTPPacket)
if packet.Seq == q.nextSeq {
if len(packet.Payload) == 0 {
q.releasePayloadMemory(packet.accountedBytes)
q.nextSeq = packet.Seq + 1
continue
}
n := copy(b, packet.Payload)
if n < len(packet.Payload) {
q.releasePayloadMemory(int64(n))
@@ -1472,8 +1602,6 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
}
if packet.Seq > q.nextSeq {
// Do not apply a packet/request count ceiling. The per-session and global
// accounted-byte budgets backpressure producers, including empty packets.
heap.Push(&q.heap, packet)
p, err := q.recv()
if err != nil {
@@ -1490,8 +1618,7 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
// A duplicate/late packet is discarded; release the bytes it owned.
q.releasePayloadMemory(packet.accountedBytes)
}
return 0, nil
}
}
func (q *nativeXHTTPUploadQueue) loadReader() io.ReadCloser {