OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

fix deadlock when changing configuration through file and API (#6077) (#6101)

Browse Source 
When changing configuration in parallel by editing the configuration
file and calling the API, the server could get into a deadlock that
prevented any further action. This is fixed.
This commit is contained in:
Alessandro Ros
2026-08-17 11:14:01 +00:00
committed by GitHub
parent 8657b48a03
commit a56c635f8e
14 changed files with 830 additions and 590 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/api.go
+7 -11
View File
@@ -7,7 +7,6 @@ import (
"net/http"
"reflect"
"sort"
"sync"
"time"
"github.com/gin-gonic/gin"
@@ -55,7 +54,13 @@ type apiAuthManager interface {
type apiParent interface {
logger.Writer
APIConfigSet(conf *conf.Conf)
APIConfigSnapshot() *conf.Conf
APIConfigGlobalPatch(conf.OptionalGlobal) error
APIConfigPathDefaultsPatch(conf.OptionalPath) error
APIConfigPathsAdd(string, conf.OptionalPath) error
APIConfigPathsPatch(string, conf.OptionalPath) error
APIConfigPathsReplace(string, conf.OptionalPath) error
APIConfigPathsDelete(string) error
}
// API is an API server.
@@ -71,7 +76,6 @@ type API struct {
TrustedProxies conf.IPNetworks
ReadTimeout conf.Duration
WriteTimeout conf.Duration
Conf *conf.Conf
AuthManager apiAuthManager
PathManager defs.APIPathManager
RTSPServer defs.APIRTSPServer
@@ -85,7 +89,6 @@ type API struct {
Parent apiParent
httpServer *httpp.Server
mutex sync.RWMutex
}
// Initialize initializes API.
@@ -287,10 +290,3 @@ func (a *API) onAuthJwksRefresh(ctx *gin.Context) {
a.AuthManager.RefreshJWTJWKS()
a.writeOK(ctx)
}
// ReloadConf is called by core.
func (a *API) ReloadConf(conf *conf.Conf) {
a.mutex.Lock()
defer a.mutex.Unlock()
a.Conf = conf
}
internal/api/api_config_global.go
+2 -17
View File
@@ -10,9 +10,7 @@ import (
)
func (a *API) onConfigGlobalGet(ctx *gin.Context) {
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
ctx.JSON(http.StatusOK, c.Global())
}
@@ -25,24 +23,11 @@ func (a *API) onConfigGlobalPatch(ctx *gin.Context) {
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.Conf.Clone()
newConf.PatchGlobal(&c)
err = newConf.Validate(nil)
err = a.Parent.APIConfigGlobalPatch(c)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
a.Conf = newConf
// since reloading the configuration can cause the shutdown of the API,
// call it in a goroutine
go a.Parent.APIConfigSet(newConf)
a.writeOK(ctx)
}
internal/api/api_config_global_test.go
+3 -6
View File
@@ -22,7 +22,6 @@ func TestConfigGlobalGet(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: &test.AuthManager{
AuthenticateImpl: func(req *auth.Request) (string, *auth.Error) {
require.Equal(t, conf.AuthActionAPI, req.Action)
@@ -32,7 +31,7 @@ func TestConfigGlobalGet(t *testing.T) {
return req.Credentials.User, nil
},
},
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -56,9 +55,8 @@ func TestConfigGlobalPatch(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -93,9 +91,8 @@ func TestConfigGlobalPatchUnknownField(t *testing.T) { //nolint:dupl
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
internal/api/api_config_pathdefaults.go
+2 -14
View File
@@ -10,9 +10,7 @@ import (
)
func (a *API) onConfigPathDefaultsGet(ctx *gin.Context) {
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
ctx.JSON(http.StatusOK, c.PathDefaults)
}
@@ -25,21 +23,11 @@ func (a *API) onConfigPathDefaultsPatch(ctx *gin.Context) {
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.Conf.Clone()
newConf.PatchPathDefaults(&p)
err = newConf.Validate(nil)
err = a.Parent.APIConfigPathDefaultsPatch(p)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
a.Conf = newConf
a.Parent.APIConfigSet(newConf)
a.writeOK(ctx)
}
internal/api/api_config_pathdefaults_test.go
+2 -4
View File
@@ -18,9 +18,8 @@ func TestConfigPathDefaultsGet(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -42,9 +41,8 @@ func TestConfigPathDefaultsPatch(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
internal/api/api_config_paths.go
+6 -66
View File
@@ -13,9 +13,7 @@ import (
)
func (a *API) onConfigPathsList(ctx *gin.Context) {
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
data := &defs.APIPathConfList{
Items: make([]conf.Path, len(c.Paths)),
@@ -43,9 +41,7 @@ func (a *API) onConfigPathsGet(ctx *gin.Context) {
return
}
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
p, ok := c.Paths[confName]
if !ok {
@@ -70,26 +66,12 @@ func (a *API) onConfigPathsAdd(ctx *gin.Context) { //nolint:dupl
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.Conf.Clone()
err = newConf.AddPath(confName, &p)
err = a.Parent.APIConfigPathsAdd(confName, p)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
err = newConf.Validate(nil)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
a.Conf = newConf
a.Parent.APIConfigSet(newConf)
a.writeOK(ctx)
}
@@ -107,12 +89,7 @@ func (a *API) onConfigPathsPatch(ctx *gin.Context) { //nolint:dupl
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.Conf.Clone()
err = newConf.PatchPath(confName, &p)
err = a.Parent.APIConfigPathsPatch(confName, p)
if err != nil {
if errors.Is(err, conf.ErrPathNotFound) {
a.writeError(ctx, http.StatusNotFound, err)
@@ -122,15 +99,6 @@ func (a *API) onConfigPathsPatch(ctx *gin.Context) { //nolint:dupl
return
}
err = newConf.Validate(nil)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
a.Conf = newConf
a.Parent.APIConfigSet(newConf)
a.writeOK(ctx)
}
@@ -148,12 +116,7 @@ func (a *API) onConfigPathsReplace(ctx *gin.Context) { //nolint:dupl
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.Conf.Clone()
err = newConf.ReplacePath(confName, &p)
err = a.Parent.APIConfigPathsReplace(confName, p)
if err != nil {
if errors.Is(err, conf.ErrPathNotFound) {
a.writeError(ctx, http.StatusNotFound, err)
@@ -163,15 +126,6 @@ func (a *API) onConfigPathsReplace(ctx *gin.Context) { //nolint:dupl
return
}
err = newConf.Validate(nil)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
a.Conf = newConf
a.Parent.APIConfigSet(newConf)
a.writeOK(ctx)
}
@@ -182,12 +136,7 @@ func (a *API) onConfigPathsDelete(ctx *gin.Context) {
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.Conf.Clone()
err := newConf.RemovePath(confName)
err := a.Parent.APIConfigPathsDelete(confName)
if err != nil {
if errors.Is(err, conf.ErrPathNotFound) {
a.writeError(ctx, http.StatusNotFound, err)
@@ -197,14 +146,5 @@ func (a *API) onConfigPathsDelete(ctx *gin.Context) {
return
}
err = newConf.Validate(nil)
if err != nil {
a.writeError(ctx, http.StatusBadRequest, err)
return
}
a.Conf = newConf
a.Parent.APIConfigSet(newConf)
a.writeOK(ctx)
}
internal/api/api_config_paths_test.go
+8 -16
View File
@@ -27,9 +27,8 @@ func TestConfigPathsList(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -70,9 +69,8 @@ func TestConfigPathsGet(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -95,9 +93,8 @@ func TestConfigPathsAdd(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -140,9 +137,8 @@ func TestConfigPathsAddUnknownField(t *testing.T) { //nolint:dupl
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -178,9 +174,8 @@ func TestConfigPathsPatch(t *testing.T) { //nolint:dupl
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -232,9 +227,8 @@ func TestConfigPathsReplace(t *testing.T) { //nolint:dupl
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -283,9 +277,8 @@ func TestConfigPathsReplaceNonExisting(t *testing.T) { //nolint:dupl
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -316,9 +309,8 @@ func TestConfigPathsDelete(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
internal/api/api_recordings.go
+3 -9
View File
@@ -57,9 +57,7 @@ func recordingsOfPath(
}
func (a *API) onRecordingsList(ctx *gin.Context) {
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
pathNames := recordstore.FindAllPathsWithSegments(c.Paths)
@@ -90,9 +88,7 @@ func (a *API) onRecordingsGet(ctx *gin.Context) {
return
}
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
pathConf, _, err := conf.FindPathConf(c.Paths, pathName)
if err != nil {
@@ -112,9 +108,7 @@ func (a *API) onRecordingDeleteSegment(ctx *gin.Context) {
return
}
a.mutex.RLock()
c := a.Conf
a.mutex.RUnlock()
c := a.Parent.APIConfigSnapshot()
pathConf, _, err := conf.FindPathConf(c.Paths, pathName)
if err != nil {
internal/api/api_recordings_test.go
+5 -10
View File
@@ -28,9 +28,8 @@ func TestRecordingsList(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -96,9 +95,8 @@ func TestRecordingsGet(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -144,9 +142,8 @@ func TestRecordingsDeleteSegment(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -196,9 +193,8 @@ func TestRecordingsDeleteSegmentInvalidPath(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
@@ -247,9 +243,8 @@ func TestRecordingsSegmentGetInvalidPath(t *testing.T) {
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
Parent: &testParent{conf: cnf},
}
err := api.Initialize()
require.NoError(t, err)
internal/api/api_test.go
+71 -9
View File
@@ -19,16 +19,84 @@ import (
)
type testParent struct {
log func(_ logger.Level, _ string, _ ...any)
log func(_ logger.Level, _ string, _ ...any)
conf *conf.Conf
}
func (p testParent) Log(l logger.Level, s string, a ...any) {
func (p *testParent) Log(l logger.Level, s string, a ...any) {
if p.log != nil {
p.log(l, s, a...)
}
}
func (testParent) APIConfigSet(_ *conf.Conf) {}
func (p *testParent) APIConfigSnapshot() *conf.Conf { return p.conf }
func (p *testParent) APIConfigGlobalPatch(in conf.OptionalGlobal) error {
newConf := p.conf.Clone()
newConf.PatchGlobal(&in)
if err := newConf.Validate(nil); err != nil {
return err
}
p.conf = newConf
return nil
}
func (p *testParent) APIConfigPathDefaultsPatch(in conf.OptionalPath) error {
newConf := p.conf.Clone()
newConf.PatchPathDefaults(&in)
if err := newConf.Validate(nil); err != nil {
return err
}
p.conf = newConf
return nil
}
func (p *testParent) APIConfigPathsAdd(name string, in conf.OptionalPath) error {
newConf := p.conf.Clone()
if err := newConf.AddPath(name, &in); err != nil {
return err
}
if err := newConf.Validate(nil); err != nil {
return err
}
p.conf = newConf
return nil
}
func (p *testParent) APIConfigPathsPatch(name string, in conf.OptionalPath) error {
newConf := p.conf.Clone()
if err := newConf.PatchPath(name, &in); err != nil {
return err
}
if err := newConf.Validate(nil); err != nil {
return err
}
p.conf = newConf
return nil
}
func (p *testParent) APIConfigPathsReplace(name string, in conf.OptionalPath) error {
newConf := p.conf.Clone()
if err := newConf.ReplacePath(name, &in); err != nil {
return err
}
if err := newConf.Validate(nil); err != nil {
return err
}
p.conf = newConf
return nil
}
func (p *testParent) APIConfigPathsDelete(name string) error {
newConf := p.conf.Clone()
if err := newConf.RemovePath(name); err != nil {
return err
}
if err := newConf.Validate(nil); err != nil {
return err
}
p.conf = newConf
return nil
}
func tempConf(t *testing.T, cnt string) *conf.Conf {
fi := test.CreateTempFile(t, []byte(cnt))
@@ -123,15 +191,12 @@ func TestPreflightRequest(t *testing.T) {
}
func TestInfo(t *testing.T) {
cnf := tempConf(t, "api: yes\n")
api := API{
Version: "v1.2.3",
Started: time.Date(2008, 11, 7, 11, 22, 0, 0, time.Local),
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: test.NilAuthManager,
Parent: &testParent{},
}
@@ -185,13 +250,10 @@ func TestAuthJWKSRefresh(t *testing.T) {
}
func TestAuthError(t *testing.T) {
cnf := tempConf(t, "api: yes\n")
api := API{
Address: "localhost:9997",
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
Conf: cnf,
AuthManager: &test.AuthManager{
AuthenticateImpl: func(req *auth.Request) (string, *auth.Error) {
if req.Credentials.User == "" {
internal/core/api_test.go
+22
View File
@@ -281,6 +281,28 @@ func TestAPIPathsList(t *testing.T) {
})
}
func TestAPIConfigGlobalPatchDisableAPI(t *testing.T) {
p, ok := newInstance(t, "api: yes\n")
require.Equal(t, true, ok)
defer p.Close()
tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPatch, "http://localhost:9997/v3/config/global/patch", map[string]any{
"api": false,
}, nil)
time.Sleep(500 * time.Millisecond)
_, err := hc.Get("http://localhost:9997/v3/config/global/get") //nolint:bodyclose
require.Error(t, err)
var urlErr *url.Error
require.ErrorAs(t, err, &urlErr)
}
func TestAPIPathsGet(t *testing.T) {
p, ok := newInstance(t, "api: yes\n"+
"paths:\n"+
internal/core/core.go
+686 -425
View File
File diff suppressed because it is too large Load Diff
internal/protocols/httpp/handler_tracker_test.go
+3 -2
View File
@@ -31,8 +31,9 @@ func TestHandlerTracker(t *testing.T) {
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
_, err2 := hc.Get("http://localhost:4667/test") //nolint:bodyclose
require.Error(t, err2)
res, err2 := hc.Get("http://localhost:4667/test")
require.NoError(t, err2)
defer res.Body.Close()
}()
<-requestReceived
internal/protocols/httpp/server.go
+10 -1
View File
@@ -2,6 +2,7 @@
package httpp
import (
"context"
"crypto/tls"
"fmt"
"log"
@@ -19,6 +20,10 @@ import (
"github.com/bluenviron/mediamtx/internal/restrictnetwork"
)
const (
shutdownTimeout = 2 * time.Second
)
type nilWriter struct{}
func (nilWriter) Write(p []byte) (int, error) {
@@ -206,7 +211,11 @@ func (s *Server) Initialize() error {
// Close closes all resources and waits for all routines to return.
func (s *Server) Close() {
s.ln.Close()
s.inner.Close() //nolint:errcheck
ctx, ctxCancel := context.WithTimeout(context.Background(), shutdownTimeout)
s.inner.Shutdown(ctx) //nolint:errcheck
ctxCancel()
s.tracker.close()
if s.loader != nil {