A stopped configuration subscriber can deadlock updates and shutdown

[YangKeao]

Bug Report

Please answer these questions before submitting your issue. Thanks!

Title: A stopped configuration subscriber can deadlock updates and shutdown

1. Minimal reproduce step (Required)

1. Start TiProxy and allow managers to register their unbuffered config watchers.
2. Cancel the server context so at least one watcher consumer exits.
3. Before shutdown completes, submit a changed configuration through the API.
4. Attempt to close the configuration manager.

2. What did you expect to see? (Required)

Stopped subscribers should be unregistered or notifications should be non-blocking and independent of the global configuration lock.

3. What did you see instead (Required)

SetTOMLConfig sends synchronously to unbuffered listeners while holding the status lock. A stopped receiver blocks the update forever and prevents Close and reads that need the same lock.

4. What is your version? (Required)

• TiProxy source commit: 51859ee68000dd14dbff4dcaf0ffaeb349d1d5be
• This report was prepared from a source audit; replace or supplement this entry with the deployed tiproxy version output when reproducing.

The following program can be used to reproduce this issue:

// Copyright 2026 PingCAP, Inc.
// SPDX-License-Identifier: Apache-2.0

package main

import (
	"bytes"
	"errors"
	"flag"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"strings"
	"syscall"
	"time"
)

type launchedProcess struct {
	cmd    *exec.Cmd
	waitCh chan error
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, "error:", err)
		os.Exit(1)
	}
}

func run() error {
	apiAddr := flag.String("api-addr", "127.0.0.1:13080", "TiProxy API address")
	pid := flag.Int("pid", 0, "existing TiProxy PID to signal; required unless --tiproxy is set")
	tiproxy := flag.String("tiproxy", "", "optional TiProxy binary to launch as a child process")
	configPath := flag.String("config", "", "config path for launched TiProxy; generated when empty")
	proxyAddr := flag.String("proxy-addr", "127.0.0.1:16000", "proxy SQL address used in generated config")
	pdAddrs := flag.String("pd-addrs", "127.0.0.1:2379", "PD address used in generated config")
	signalName := flag.String("signal", "TERM", "signal used to cancel TiProxy: TERM or INT")
	delayAfterSignal := flag.Duration("delay-after-signal", 50*time.Millisecond, "delay between signal and config PUT")
	startupTimeout := flag.Duration("startup-timeout", 15*time.Second, "time to wait for launched TiProxy API")
	putTimeout := flag.Duration("put-timeout", 5*time.Second, "config PUT timeout")
	exitTimeout := flag.Duration("exit-timeout", 10*time.Second, "time to wait for process exit after PUT")
	killAfterTimeout := flag.Bool("kill-after-timeout", false, "SIGKILL launched child if it does not exit by --exit-timeout")
	putBody := flag.String("put-body", "[log]\nlevel = \"debug\"\n", "TOML body sent to /api/admin/config/")
	flag.Parse()

	sig, err := parseSignal(*signalName)
	if err != nil {
		return err
	}

	var child *launchedProcess
	targetPID := *pid
	if *tiproxy != "" {
		child, err = launchTiProxy(*tiproxy, *configPath, *proxyAddr, *apiAddr, *pdAddrs, *startupTimeout)
		if err != nil {
			return err
		}
		targetPID = child.cmd.Process.Pid
		fmt.Printf("launched TiProxy PID %d\n", targetPID)
	} else if targetPID <= 0 {
		return errors.New("set either --pid for an existing TiProxy or --tiproxy to launch one")
	}

	if err := signalProcess(targetPID, sig); err != nil {
		return err
	}
	fmt.Printf("sent %s to PID %d; waiting %s before config PUT\n", sig, targetPID, *delayAfterSignal)
	time.Sleep(*delayAfterSignal)

	start := time.Now()
	err = putConfig(*apiAddr, []byte(*putBody), *putTimeout)
	elapsed := time.Since(start)
	if err != nil {
		fmt.Printf("config PUT returned after %s with error: %v\n", elapsed.Round(time.Millisecond), err)
	} else {
		fmt.Printf("config PUT completed in %s\n", elapsed.Round(time.Millisecond))
	}

	if child != nil {
		return waitLaunched(child, *exitTimeout, *killAfterTimeout)
	}
	return waitExisting(targetPID, *exitTimeout)
}

func launchTiProxy(binaryPath, configPath, proxyAddr, apiAddr, pdAddrs string, startupTimeout time.Duration) (*launchedProcess, error) {
	var err error
	if configPath == "" {
		configPath, err = writeGeneratedConfig(proxyAddr, apiAddr, pdAddrs)
		if err != nil {
			return nil, err
		}
		fmt.Printf("generated config %s\n", configPath)
	}

	cmd := exec.Command(binaryPath, "--config", configPath)
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("start %s: %w", binaryPath, err)
	}
	lp := &launchedProcess{cmd: cmd, waitCh: make(chan error, 1)}
	go func() {
		lp.waitCh <- cmd.Wait()
	}()

	if err := waitAPI(apiAddr, startupTimeout); err != nil {
		_ = cmd.Process.Kill()
		<-lp.waitCh
		return nil, err
	}
	return lp, nil
}

func writeGeneratedConfig(proxyAddr, apiAddr, pdAddrs string) (string, error) {
	dir, err := os.MkdirTemp("", "tiproxy-medium-003-")
	if err != nil {
		return "", fmt.Errorf("create temp dir: %w", err)
	}
	workdir := dir + "/work"
	if err := os.MkdirAll(workdir, 0o755); err != nil {
		return "", fmt.Errorf("create workdir: %w", err)
	}
	config := fmt.Sprintf(`workdir = %q

[proxy]
addr = %q
pd-addrs = %q
graceful-close-conn-timeout = 60

[api]
addr = %q

[log]
level = "info"
encoder = "tidb"
`, workdir, proxyAddr, pdAddrs, apiAddr)

	path := dir + "/tiproxy.toml"
	if err := os.WriteFile(path, []byte(config), 0o644); err != nil {
		return "", fmt.Errorf("write generated config: %w", err)
	}
	return path, nil
}

func waitAPI(apiAddr string, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	url := apiURL(apiAddr)
	client := &http.Client{Timeout: 500 * time.Millisecond}
	var lastErr error
	for time.Now().Before(deadline) {
		resp, err := client.Get(url)
		if err == nil {
			_, _ = io.Copy(io.Discard, resp.Body)
			_ = resp.Body.Close()
			if resp.StatusCode >= 200 && resp.StatusCode < 500 {
				fmt.Printf("API is reachable at %s\n", url)
				return nil
			}
			lastErr = fmt.Errorf("GET %s status %s", url, resp.Status)
		} else {
			lastErr = err
		}
		time.Sleep(100 * time.Millisecond)
	}
	return fmt.Errorf("API did not become reachable within %s: %w", timeout, lastErr)
}

func putConfig(apiAddr string, body []byte, timeout time.Duration) error {
	req, err := http.NewRequest(http.MethodPut, apiURL(apiAddr), bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/toml")
	client := &http.Client{Timeout: timeout}
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("status %s: %s", resp.Status, strings.TrimSpace(string(respBody)))
	}
	return nil
}

func apiURL(apiAddr string) string {
	base := apiAddr
	if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") {
		base = "http://" + base
	}
	return strings.TrimRight(base, "/") + "/api/admin/config/"
}

func parseSignal(name string) (syscall.Signal, error) {
	switch strings.ToUpper(strings.TrimPrefix(name, "SIG")) {
	case "TERM":
		return syscall.SIGTERM, nil
	case "INT":
		return syscall.SIGINT, nil
	default:
		return 0, fmt.Errorf("unsupported --signal %q", name)
	}
}

func signalProcess(pid int, sig syscall.Signal) error {
	if pid <= 0 {
		return fmt.Errorf("invalid pid %d", pid)
	}
	if err := syscall.Kill(pid, sig); err != nil {
		return fmt.Errorf("send %s to pid %d: %w", sig, pid, err)
	}
	return nil
}

func waitLaunched(child *launchedProcess, timeout time.Duration, killAfterTimeout bool) error {
	select {
	case err := <-child.waitCh:
		if err != nil {
			fmt.Printf("launched TiProxy exited with: %v\n", err)
		} else {
			fmt.Println("launched TiProxy exited cleanly")
		}
		return nil
	case <-time.After(timeout):
		fmt.Printf("launched TiProxy did not exit within %s\n", timeout)
		if killAfterTimeout {
			fmt.Println("killing launched TiProxy child")
			_ = child.cmd.Process.Kill()
			<-child.waitCh
		}
		return nil
	}
}

func waitExisting(pid int, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if !processExists(pid) {
			fmt.Printf("PID %d exited\n", pid)
			return nil
		}
		time.Sleep(100 * time.Millisecond)
	}
	fmt.Printf("PID %d still exists after %s\n", pid, timeout)
	return nil
}

func processExists(pid int) bool {
	err := syscall.Kill(pid, 0)
	return err == nil || errors.Is(err, syscall.EPERM)
}

[YangKeao]

/assign @YangKeao