Github-Mirror/pocketbase
1
0
Fork 0
mirror of https://github.com/pocketbase/pocketbase.git synced 2026-09-08 23:50:47 +02:00
Code Issues Packages Projects Releases Wiki Activity

[#7836] fixed migration logs write deadlock and added app.ClearBootstrap/OnClearBootstrap helpers

Browse Source
This commit is contained in:
Gani Georgiev
2026-09-07 19:31:34 +03:00
parent 5684ee24f1
commit 114c01ac12
18 changed files with 3790 additions and 3576 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/app.go
+11 -2
View File
@@ -54,11 +54,16 @@ type App interface {
// Bootstrap initializes the application
// (aka. create data dir, open db connections, load settings, etc.).
//
// It will call ResetBootstrapState() if the application was already bootstrapped.
// It calls ClearBootstrap() if the application was already bootstrapped.
Bootstrap() error
// ResetBootstrapState releases the initialized core app resources
// ClearBootstrap releases the initialized core app resources
// (closing db connections, stopping cron ticker, etc.).
//
// This method is no-op if the application is not bootstrapped yet.
ClearBootstrap() error
// Deprecated: use ClearBootstrap().
ResetBootstrapState() error
// DataDir returns the app data directory path.
@@ -713,6 +718,10 @@ type App interface {
// resources (db, app settings, etc).
OnBootstrap() *hook.Hook[*BootstrapEvent]
// OnClearBootstrap hook is triggered on unsetting the main application
// resources (db, app settings, etc) to their initial nil/empty value.
OnClearBootstrap() *hook.Hook[*BootstrapEvent]
// OnServe hook is triggered when the app web server is started
// (after starting the TCP listener but before initializing the blocking serve task),
// allowing you to adjust its options and attach new routes or middlewares.
core/base.go
+120 -65
View File
@@ -12,6 +12,7 @@ import (
"regexp"
"runtime"
"strings"
"sync/atomic"
"time"
"github.com/fatih/color"
@@ -89,11 +90,12 @@ type BaseApp struct {
auxNonconcurrentDB dbx.Builder
// app event hooks
onBootstrap *hook.Hook[*BootstrapEvent]
onServe *hook.Hook[*ServeEvent]
onTerminate *hook.Hook[*TerminateEvent]
onBackupCreate *hook.Hook[*BackupEvent]
onBackupRestore *hook.Hook[*BackupEvent]
onBootstrap *hook.Hook[*BootstrapEvent]
onClearBootstrap *hook.Hook[*BootstrapEvent]
onServe *hook.Hook[*ServeEvent]
onTerminate *hook.Hook[*TerminateEvent]
onBackupCreate *hook.Hook[*BackupEvent]
onBackupRestore *hook.Hook[*BackupEvent]
// db model hooks
onModelValidate *hook.Hook[*ModelEvent]
@@ -249,6 +251,7 @@ func NewBaseApp(config BaseAppConfig) *BaseApp {
func (app *BaseApp) initHooks() {
// app event hooks
app.onBootstrap = &hook.Hook[*BootstrapEvent]{}
app.onClearBootstrap = &hook.Hook[*BootstrapEvent]{}
app.onServe = &hook.Hook[*ServeEvent]{}
app.onTerminate = &hook.Hook[*TerminateEvent]{}
app.onBackupCreate = &hook.Hook[*BackupEvent]{}
@@ -405,14 +408,14 @@ func (app *BaseApp) IsBootstrapped() bool {
// Bootstrap initializes the application
// (aka. create data dir, open db connections, load settings, etc.).
//
// It will call ResetBootstrapState() if the application was already bootstrapped.
// It calls ClearBootstrap() if the application was already bootstrapped.
func (app *BaseApp) Bootstrap() error {
event := &BootstrapEvent{}
event.App = app
err := app.OnBootstrap().Trigger(event, func(e *BootstrapEvent) error {
// clear resources of previous core state (if any)
if err := app.ResetBootstrapState(); err != nil {
// clear previous bootstrap state (if any)
if err := app.ClearBootstrap(); err != nil {
return err
}
@@ -460,41 +463,55 @@ func (app *BaseApp) Bootstrap() error {
return err
}
type closer interface {
Close() error
// Deprecated: use [ClearBootstrap].
func (app *BaseApp) ResetBootstrapState() error {
return app.ClearBootstrap()
}
// ResetBootstrapState releases the initialized core app resources
// ClearBootstrap releases the initialized core app resources
// (closing db connections, stopping cron ticker, etc.).
func (app *BaseApp) ResetBootstrapState() error {
app.Cron().Stop()
var errs []error
dbs := []*dbx.Builder{
&app.concurrentDB,
&app.nonconcurrentDB,
&app.auxConcurrentDB,
&app.auxNonconcurrentDB,
//
// This method is no-op if the application is not bootstrapped yet.
func (app *BaseApp) ClearBootstrap() error {
if !app.IsBootstrapped() {
return nil
}
for _, db := range dbs {
if db == nil {
continue
event := &BootstrapEvent{}
event.App = app
return app.OnClearBootstrap().Trigger(event, func(e *BootstrapEvent) error {
type closer interface {
Close() error
}
if v, ok := (*db).(closer); ok {
if err := v.Close(); err != nil {
errs = append(errs, err)
var errs []error
dbs := []*dbx.Builder{
&app.concurrentDB,
&app.nonconcurrentDB,
&app.auxConcurrentDB,
&app.auxNonconcurrentDB,
}
for _, db := range dbs {
if db == nil {
continue
}
if v, ok := (*db).(closer); ok {
if err := v.Close(); err != nil {
errs = append(errs, err)
}
}
*db = nil
}
*db = nil
}
if len(errs) > 0 {
return errors.Join(errs...)
}
if len(errs) > 0 {
return errors.Join(errs...)
}
return nil
return nil
})
}
// DB returns the default app data.db builder instance.
@@ -817,7 +834,7 @@ func (app *BaseApp) Restart() error {
event.IsRestart = true
return app.OnTerminate().Trigger(event, func(e *TerminateEvent) error {
_ = e.App.ResetBootstrapState()
_ = e.App.ClearBootstrap()
// attempt to restart the bootstrap process in case execve returns an error for some reason
defer func() {
@@ -860,6 +877,10 @@ func (app *BaseApp) OnBootstrap() *hook.Hook[*BootstrapEvent] {
return app.onBootstrap
}
func (app *BaseApp) OnClearBootstrap() *hook.Hook[*BootstrapEvent] {
return app.onClearBootstrap
}
func (app *BaseApp) OnServe() *hook.Hook[*ServeEvent] {
return app.onServe
}
@@ -1413,7 +1434,15 @@ func (app *BaseApp) registerBaseHooks() {
Id: "__pbCronStart__",
Func: func(e *ServeEvent) error {
app.Cron().Start()
return e.Next()
},
Priority: 999,
})
app.OnClearBootstrap().Bind(&hook.Handler[*BootstrapEvent]{
Id: "__pbCronStop__",
Func: func(e *BootstrapEvent) error {
app.Cron().Stop()
return e.Next()
},
Priority: 999,
@@ -1470,9 +1499,41 @@ func getLoggerMinLevel(app App) slog.Level {
}
func (app *BaseApp) initLogger() error {
var stopped atomic.Bool
duration := 3 * time.Second
ticker := time.NewTicker(duration)
done := make(chan bool, 1)
done := make(chan struct{}, 1)
runLogsWrite := func(logs []*logger.Log) {
if !app.IsBootstrapped() || app.Settings().Logs.MaxDays == 0 {
return
}
// write the accumulated logs
//
// note: based on several local tests there is no
// significant performance difference between small number
// of separate write queries vs 1 big INSERT
app.AuxRunInTransaction(func(txApp App) error {
model := &Log{}
for _, l := range logs {
model.MarkAsNew()
model.Id = GenerateDefaultRandomId()
model.Level = int(l.Level)
model.Message = l.Message
model.Data = l.Data
model.Created, _ = types.ParseDateTime(l.Time)
if err := txApp.AuxSave(model); err != nil {
log.Println("Failed to write log", model, err)
}
}
return nil
})
}
handler := logger.NewBatchHandler(logger.BatchOptions{
Level: getLoggerMinLevel(app),
@@ -1487,35 +1548,25 @@ func (app *BaseApp) initLogger() error {
}
}
ticker.Reset(duration)
if !stopped.Load() {
ticker.Reset(duration)
}
return app.Settings().Logs.MaxDays > 0
},
WriteFunc: func(ctx context.Context, logs []*logger.Log) error {
if !app.IsBootstrapped() || app.Settings().Logs.MaxDays == 0 {
return nil
// don't block and wait for the write transaction to complete
// when we can't be sure if the logs write wasn't triggered while
// inside another AUX db transaction (ticker or batch threshold reached)
// which can block indefinitely and cause deadlock
// (https://github.com/pocketbase/pocketbase/issues/7836)
shouldBlock, _ := ctx.Value(logger.BlockKey).(bool)
if shouldBlock {
runLogsWrite(logs)
} else {
routine.FireAndForget(func() { runLogsWrite(logs) })
}
// write the accumulated logs
// (note: based on several local tests there is no significant performance difference between small number of separate write queries vs 1 big INSERT)
app.AuxRunInTransaction(func(txApp App) error {
model := &Log{}
for _, l := range logs {
model.MarkAsNew()
model.Id = GenerateDefaultRandomId()
model.Level = int(l.Level)
model.Message = l.Message
model.Data = l.Data
model.Created, _ = types.ParseDateTime(l.Time)
if err := txApp.AuxSave(model); err != nil {
log.Println("Failed to write log", model, err)
}
}
return nil
})
return nil
},
})
@@ -1535,17 +1586,21 @@ func (app *BaseApp) initLogger() error {
app.logger = slog.New(handler)
// write all remaining logs before ticker.Stop to avoid races with ResetBootstrap user calls
app.OnTerminate().Bind(&hook.Handler[*TerminateEvent]{
Id: "__pbAppLoggerOnTerminate__",
Func: func(e *TerminateEvent) error {
handler.WriteAll(context.Background())
// attempt to write all queued logs before clearing the application bootstrap state
app.OnClearBootstrap().Bind(&hook.Handler[*BootstrapEvent]{
Id: "__pbAppLoggerFlushBeforeStop__",
Func: func(e *BootstrapEvent) error {
// extra precaution in case the hook was manually triggered while inside aux db transaction
_, isTx := e.App.AuxNonconcurrentDB().(*dbx.Tx)
ctx := context.WithValue(context.Background(), logger.BlockKey, !isTx)
handler.WriteAll(ctx)
stopped.Store(true)
ticker.Stop()
// don't block in case OnTerminate is triggered more than once
// don't block in case the hook is triggered more than once
select {
case done <- true:
case done <- struct{}{}:
default:
}
core/base_test.go
+144 -43
View File
@@ -65,7 +65,7 @@ func TestBaseAppBootstrap(t *testing.T) {
app := core.NewBaseApp(core.BaseAppConfig{
DataDir: testDataDir,
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
if app.IsBootstrapped() {
t.Fatal("Didn't expect the application to be bootstrapped.")
@@ -115,7 +115,7 @@ func TestBaseAppBootstrap(t *testing.T) {
runNilChecks(nilChecksBeforeReset)
// reset
if err := app.ResetBootstrapState(); err != nil {
if err := app.ClearBootstrap(); err != nil {
t.Fatal(err)
}
@@ -141,7 +141,7 @@ func TestNewBaseAppTx(t *testing.T) {
app := core.NewBaseApp(core.BaseAppConfig{
DataDir: testDataDir,
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
if err := app.Bootstrap(); err != nil {
t.Fatal(err)
@@ -185,7 +185,7 @@ func TestBaseAppNewMailClient(t *testing.T) {
DataDir: testDataDir,
EncryptionEnv: "pb_test_env",
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
client1 := app.NewMailClient()
m1, ok := client1.(*mailer.Sendmail)
@@ -215,7 +215,7 @@ func TestBaseAppNewFilesystem(t *testing.T) {
app := core.NewBaseApp(core.BaseAppConfig{
DataDir: testDataDir,
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
// local
local, localErr := app.NewFilesystem()
@@ -244,7 +244,7 @@ func TestBaseAppNewBackupsFilesystem(t *testing.T) {
app := core.NewBaseApp(core.BaseAppConfig{
DataDir: testDataDir,
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
// local
local, localErr := app.NewBackupsFilesystem()
@@ -266,74 +266,175 @@ func TestBaseAppNewBackupsFilesystem(t *testing.T) {
}
}
const logsThreshold = 200
func assertLogsCount(t *testing.T, app core.App, expected int) {
var total int
err := app.LogQuery().Select("count(*)").Row(&total)
if err != nil {
t.Fatalf("Failed to fetch total logs: %v", err)
}
if total != expected {
t.Fatalf("Expected %d log(s), got %d", expected, total)
}
}
func TestBaseAppLoggerWrites(t *testing.T) {
t.Parallel()
// note: outside of synctest because the bootstrap tickers could deadlock
app, _ := tests.NewTestApp()
defer app.Cleanup()
// reset
if err := app.DeleteOldLogs(time.Now()); err != nil {
// clear old logs
err := app.DeleteOldLogs(time.Now())
if err != nil {
t.Fatal(err)
}
const logsThreshold = 200
totalLogs := func(app core.App, t *testing.T) int {
var total int
err := app.LogQuery().Select("count(*)").Row(&total)
if err != nil {
t.Fatalf("Failed to fetch total logs: %v", err)
}
return total
}
t.Run("disabled logs retention", func(t *testing.T) {
app.Settings().Logs.MaxDays = 0
synctest.Test(t, func(t *testing.T) {
app.Settings().Logs.MaxDays = 0
for i := 0; i < logsThreshold+1; i++ {
app.Logger().Error("test")
}
for i := 0; i < logsThreshold+1; i++ {
app.Logger().Error("test")
}
if total := totalLogs(app, t); total != 0 {
t.Fatalf("Expected no logs, got %d", total)
}
// short delay for the non-blocking write goroutine
synctest.Sleep(time.Nanosecond)
assertLogsCount(t, app, 0)
})
})
t.Run("test batch logs writes", func(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
app.Settings().Logs.MaxDays = 1
app.Settings().Logs.MaxDays = 2
for i := 0; i < logsThreshold-1; i++ {
app.Logger().Error("test")
}
if total := totalLogs(app, t); total != 0 {
t.Fatalf("Expected no logs, got %d", total)
}
// short delay for the non-blocking write goroutine
synctest.Sleep(time.Nanosecond)
// should trigger batch write
// below threshold
assertLogsCount(t, app, 0)
// threshold reached -> should trigger batch write
app.Logger().Error("test")
// should be added for the next batch write
// should be skipped from this batch and added for the next
app.Logger().Error("test")
if total := totalLogs(app, t); total != logsThreshold {
t.Fatalf("Expected %d logs, got %d", logsThreshold, total)
}
// short delay for the non-blocking write goroutine
synctest.Sleep(time.Nanosecond)
// wait for 3 secs to check the timer trigger
synctest.Sleep(3000 * time.Millisecond)
assertLogsCount(t, app, logsThreshold)
if total := totalLogs(app, t); total != logsThreshold {
t.Fatalf("Expected %d logs, got %d", logsThreshold, total)
}
// note: we can't test the flush timer here because the ticker
// was started out of the synctest buble to avoid deadlocks
// (see TestBaseAppLoggerWritesAwaited for a flaky but real timer test)
})
})
}
func TestBaseAppLoggerWritesAwaited(t *testing.T) {
t.Parallel()
app, _ := tests.NewTestApp()
defer app.Cleanup()
// clear old logs
err := app.DeleteOldLogs(time.Now())
if err != nil {
t.Fatal(err)
}
// enable logs persistence
app.Settings().Logs.MaxDays = 1
err = app.Save(app.Settings())
if err != nil {
t.Fatal(err)
}
t.Run("flush on timer tick", func(t *testing.T) {
timeout := time.After(5 * time.Second)
done := make(chan struct{})
logsHook := app.OnModelAfterCreateSuccess("_logs")
hookId := logsHook.BindFunc(func(e *core.ModelEvent) error {
done <- struct{}{}
return e.Next()
})
defer logsHook.Unbind(hookId)
app.Logger().Error("test")
// short wait to ensure that there is no non-blocking write
time.Sleep(500 * time.Millisecond)
assertLogsCount(t, app, 0)
// wait for the ticker to write the db record
select {
case <-timeout:
t.Fatal("ticker wait timeout")
case <-done:
}
assertLogsCount(t, app, 1)
})
t.Run("before ClearBootstrap flush", func(t *testing.T) {
app.Logger().Error("test")
app.Bootstrap()
assertLogsCount(t, app, 2)
})
t.Run("batch flush inside aux transaction shouldn't hang", func(t *testing.T) {
timeout := time.After(1 * time.Second)
done := make(chan struct{})
totalCreated := 0
logsHook := app.OnModelAfterCreateSuccess("_logs")
hookId := logsHook.BindFunc(func(e *core.ModelEvent) error {
totalCreated++
if totalCreated == 200 {
done <- struct{}{}
}
return e.Next()
})
defer logsHook.Unbind(hookId)
app.AuxRunInTransaction(func(txApp core.App) error {
for range logsThreshold {
txApp.Logger().Error("test")
}
return nil
})
// wait for the non-blocking write
select {
case <-timeout:
t.Fatal("non-blocking write timeout")
case <-done:
}
assertLogsCount(t, app, 202)
// force clear to ensure that there are no other logs
app.Bootstrap()
assertLogsCount(t, app, 202)
})
}
func TestBaseAppRefreshSettingsLoggerMinLevelEnabled(t *testing.T) {
scenarios := []struct {
name string
@@ -373,7 +474,7 @@ func TestBaseAppRefreshSettingsLoggerMinLevelEnabled(t *testing.T) {
DataDir: testDataDir,
IsDev: s.isDev,
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
if err := app.Bootstrap(); err != nil {
t.Fatal(err)
core/log_printer_test.go
+9 -2
View File
@@ -3,17 +3,24 @@ package core
import (
"context"
"database/sql"
"io"
"log/slog"
"os"
"testing"
"time"
"github.com/fatih/color"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/tools/list"
"github.com/pocketbase/pocketbase/tools/logger"
)
func TestBaseAppLoggerLevelDevPrint(t *testing.T) {
// temp unset to avoid littering the stdout if the test fails when in dev mode
colorOutput := color.Output
color.Output = io.Discard
defer func() { color.Output = colorOutput }()
testLogLevel := 4
scenarios := []struct {
@@ -48,7 +55,7 @@ func TestBaseAppLoggerLevelDevPrint(t *testing.T) {
DataDir: testDataDir,
IsDev: s.isDev,
})
defer app.ResetBootstrapState()
defer app.ClearBootstrap()
if err := app.Bootstrap(); err != nil {
t.Fatal(err)
@@ -68,7 +75,7 @@ func TestBaseAppLoggerLevelDevPrint(t *testing.T) {
var printedLevels []int
var persistedLevels []int
ctx := context.Background()
ctx := context.WithValue(context.Background(), logger.BlockKey, true)
// track printed logs
originalPrintLog := printLog
core/system_alert_test.go
+2 -2
View File
@@ -19,7 +19,7 @@ func TestSendSystemAlert(t *testing.T) {
testApp := NewBaseApp(BaseAppConfig{
DataDir: testDataDir,
})
defer testApp.ResetBootstrapState()
defer testApp.ClearBootstrap()
if err := testApp.Bootstrap(); err != nil {
t.Fatal(err)
@@ -72,7 +72,7 @@ func TestSendSystemAlertToAllSuperusers(t *testing.T) {
testApp := NewBaseApp(BaseAppConfig{
DataDir: testDataDir,
})
defer testApp.ResetBootstrapState()
defer testApp.ClearBootstrap()
if err := testApp.Bootstrap(); err != nil {
t.Fatal(err)