Improve error handling on AWS DB failover (#39841)

<!-- Add the related story/sub-task/bug number, like Resolves #123, or
remove if NA -->
**Related issue:** Resolves #39228 

Manually tested by triggering a failover on loadtest.

# Checklist for submitter

If some of the following don't apply, delete the relevant line.

- [x] Changes file added for user-visible changes in `changes/`,
`orbit/changes/` or `ee/fleetd-chrome/changes`.

## Testing

- [x] Added/updated automated tests
- [x] QA'd all new/changed functionality manually


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Health checks now detect a primary DB becoming read-only and report
failure so the service restarts and reconnects to a writable primary.
* Write failures due to DB read-only state now trigger immediate fatal
handling to prompt graceful shutdown and recovery.
* Improved detection and handling of read-only DB conditions to increase
stability during failovers.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

changes/39228-db-failover

@@ -0,0 +1 @@
+Improve error handling on AWS DB failover. Fleet will now fail health check if the primary DB is read-only, or trigger graceful shutdown when write operations encounter read-only errors.

cmd/fleet/serve.go

@@ -823,6 +823,16 @@ the way that the Fleet server works.
 			ctx, cancelFunc := context.WithCancel(baseCtx)
 			defer cancelFunc()
 
+			// Channel used to trigger graceful shutdown on fatal DB errors (e.g. Aurora failover).
+			dbFatalCh := make(chan error, 1)
+			common_mysql.SetFatalErrorHandler(func(err error) {
+				level.Error(logger).Log("msg", "fatal database error detected, initiating graceful shutdown", "err", err)
+				select {
+				case dbFatalCh <- err:
+				default:
+				}
+			})
+
 			var conditionalAccessMicrosoftProxy *conditional_access_microsoft_proxy.Proxy
 			if config.MicrosoftCompliancePartner.IsSet() {
 				var err error
@@ -1718,7 +1728,10 @@ the way that the Fleet server works.
 			go func() {
 				sig := make(chan os.Signal, 1)
 				signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
-				<-sig // block on signal
+				select {
+				case <-sig:
+				case <-dbFatalCh:
+				}
 				ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 				defer cancel()
 				errs <- func() error {

server/datastore/mysql/mysql.go

@@ -731,9 +731,20 @@ func (ds *Datastore) HealthCheck() error {
 	// NOTE: does not receive a context as argument here, because the HealthCheck
 	// interface potentially affects more than the datastore layer, and I'm not
 	// sure we can safely identify and change them all at this moment.
-	if _, err := ds.primary.ExecContext(context.Background(), "select 1"); err != nil {
+
+	// Check that the primary is reachable and not in read-only mode.
+	// After an AWS Aurora failover the old writer is demoted to a reader;
+	// detecting this lets the health check fail so the orchestrator can restart Fleet.
+	var readOnly int
+	if err := ds.primary.QueryRowContext(context.Background(), "SELECT @@read_only").Scan(&readOnly); err != nil {
 		return err
 	}
+	if readOnly == 1 {
+		// Intentionally return an error so that the health check endpoint returns a 500,
+		// signaling the orchestrator (ECS, Kubernetes) to restart Fleet with fresh DB connections.
+		return errors.New("primary database is read-only, possible failover detected")
+	}
+
 	if ds.readReplicaConfig != nil {
 		var dst int
 		if err := sqlx.GetContext(context.Background(), ds.replica, &dst, "select 1"); err != nil {

server/datastore/mysql/mysql_test.go

@@ -702,6 +702,25 @@ func TestWithTxWillRollbackWhenPanic(t *testing.T) {
 	require.NoError(t, mock.ExpectationsWereMet())
 }
 
+func TestHealthCheckDetectsReadOnly(t *testing.T) {
+	mock, ds := mockDatastore(t)
+	defer ds.Close()
+
+	// Healthy: primary is writable.
+	mock.ExpectQuery("SELECT @@read_only").
+		WillReturnRows(sqlmock.NewRows([]string{"@@read_only"}).AddRow(0))
+	require.NoError(t, ds.HealthCheck())
+
+	// Unhealthy: primary is read-only (failover scenario).
+	mock.ExpectQuery("SELECT @@read_only").
+		WillReturnRows(sqlmock.NewRows([]string{"@@read_only"}).AddRow(1))
+	err := ds.HealthCheck()
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "read-only")
+
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
 func TestNewReadsPasswordFromDisk(t *testing.T) {
 	passwordFile, err := os.CreateTemp(t.TempDir(), "*.passwordtest")
 	require.NoError(t, err)

server/datastore/mysql/sessions.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+
 	"github.com/fleetdm/fleet/v4/server"
 	"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
 	"github.com/fleetdm/fleet/v4/server/fleet"
+	common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
 	"github.com/jmoiron/sqlx"
 )
 
@@ -188,6 +190,9 @@ func (ds *Datastore) MarkSessionAccessed(ctx context.Context, session *fleet.Ses
 	`
 	results, err := ds.writer(ctx).ExecContext(ctx, sqlStatement, ds.clock.Now(), session.ID)
 	if err != nil {
+		if common_mysql.IsReadOnlyError(err) {
+			common_mysql.TriggerFatalError(err)
+		}
 		return ctxerr.Wrap(ctx, err, "updating mark session as accessed")
 	}
 	rows, err := results.RowsAffected()

server/platform/mysql/common.go

@@ -180,11 +180,18 @@ func WithTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) error
 		if rbErr != nil && rbErr != sql.ErrTxDone {
 			return ctxerr.Wrapf(ctx, err, "got err '%s' rolling back after err", rbErr.Error())
 		}
+		if IsReadOnlyError(err) {
+			TriggerFatalError(err)
+		}
 		return err
 	}
 
 	if err := tx.Commit(); err != nil {
-		return ctxerr.Wrap(ctx, err, "commit transaction")
+		err = ctxerr.Wrap(ctx, err, "commit transaction")
+		if IsReadOnlyError(err) {
+			TriggerFatalError(err)
+		}
+		return err
 	}
 
 	return nil

server/platform/mysql/errors.go

@@ -2,9 +2,12 @@ package mysql
 
 import (
 	"database/sql"
+	"errors"
 	"fmt"
 
+	"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
 	platform_http "github.com/fleetdm/fleet/v4/server/platform/http"
+	"github.com/go-sql-driver/mysql"
 )
 
 type NotFoundError struct {
@@ -66,3 +69,29 @@ func (e *NotFoundError) IsClientError() bool {
 func (e *NotFoundError) Is(other error) bool {
 	return other == sql.ErrNoRows
 }
+
+// MySQL error numbers for read-only conditions. These are not included in the
+// VividCortex/mysqlerr package, so we define them here.
+const (
+	// erReadOnlyTransaction is MySQL error 1792: Cannot execute statement in a READ ONLY transaction.
+	erReadOnlyTransaction = 1792
+	// erOptionPreventsStatement is MySQL error 1290: The MySQL server is running with the --read-only option.
+	erOptionPreventsStatement = 1290
+	// erReadOnlyMode is MySQL error 1836: Running in read-only mode.
+	erReadOnlyMode = 1836
+)
+
+// IsReadOnlyError returns true if the error is a MySQL error indicating that
+// the server is in read-only mode. This typically happens after an Aurora
+// failover when the primary has been demoted to a reader.
+func IsReadOnlyError(err error) bool {
+	err = ctxerr.Cause(err)
+	var mySQLErr *mysql.MySQLError
+	if errors.As(err, &mySQLErr) {
+		switch mySQLErr.Number {
+		case erReadOnlyTransaction, erOptionPreventsStatement, erReadOnlyMode:
+			return true
+		}
+	}
+	return false
+}

server/platform/mysql/errors_test.go

@@ -0,0 +1,52 @@
+package mysql
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsReadOnlyError(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "unrelated MySQL error",
+			err:  &mysql.MySQLError{Number: 1045, Message: "Access denied"},
+			want: false,
+		},
+		{
+			name: "error 1792 read-only transaction",
+			err:  &mysql.MySQLError{Number: 1792, Message: "Cannot execute statement in a READ ONLY transaction."},
+			want: true,
+		},
+		{
+			name: "error 1290 option prevents statement",
+			err:  &mysql.MySQLError{Number: 1290, Message: "The MySQL server is running with the --read-only option"},
+			want: true,
+		},
+		{
+			name: "error 1836 read-only mode",
+			err:  &mysql.MySQLError{Number: 1836, Message: "Running in read-only mode"},
+			want: true,
+		},
+		{
+			name: "wrapped read-only error",
+			err:  fmt.Errorf("transaction failed: %w", &mysql.MySQLError{Number: 1792, Message: "read only"}),
+			want: true,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, IsReadOnlyError(tc.err))
+		})
+	}
+}

server/platform/mysql/retry.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
+	"sync"
 	"time"
 
 	"github.com/VividCortex/mysqlerr"
@@ -14,6 +16,40 @@ import (
 	"github.com/jmoiron/sqlx"
 )
 
+var (
+	fatalErrorMu      sync.RWMutex
+	fatalErrorHandler func(error)
+	fatalErrorOnce    sync.Once
+)
+
+// SetFatalErrorHandler registers a function that will be called (at most once)
+// when a fatal database error is detected, such as the primary becoming
+// read-only during an Aurora failover. The handler should trigger a graceful
+// process shutdown.
+//
+// If no handler is set, the default behavior is to panic.
+func SetFatalErrorHandler(fn func(error)) {
+	fatalErrorMu.Lock()
+	defer fatalErrorMu.Unlock()
+	fatalErrorHandler = fn
+	fatalErrorOnce = sync.Once{} // reset so handler fires on next fatal error
+}
+
+// TriggerFatalError calls the registered fatal error handler exactly once.
+// If no handler is registered, it panics (legacy behavior).
+func TriggerFatalError(err error) {
+	fatalErrorMu.RLock()
+	defer fatalErrorMu.RUnlock()
+
+	if fatalErrorHandler == nil {
+		panic(fmt.Sprintf("database is read-only, possible failover detected: %v", err))
+	}
+
+	fatalErrorOnce.Do(func() {
+		fatalErrorHandler(err)
+	})
+}
+
 var DoRetryErr = errors.New("fleet datastore retry")
 
 type TxFn func(tx sqlx.ExtContext) error
@@ -45,6 +81,13 @@ func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger)
 				return backoff.Permanent(ctxerr.Wrapf(ctx, err, "got err '%s' rolling back after err", rbErr.Error()))
 			}
 
+			// Read-only errors indicate a DB failover occurred (primary demoted to reader).
+			// Trigger graceful shutdown so the orchestrator restarts and reconnects to the new primary.
+			if IsReadOnlyError(err) {
+				TriggerFatalError(err)
+				return backoff.Permanent(err)
+			}
+
 			if retryableError(err) {
 				return err
 			}
@@ -56,6 +99,11 @@ func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger)
 		if err := tx.Commit(); err != nil {
 			err = ctxerr.Wrap(ctx, err, "commit transaction")
 
+			if IsReadOnlyError(err) {
+				TriggerFatalError(err)
+				return backoff.Permanent(err)
+			}
+
 			if retryableError(err) {
 				return err
 			}

server/platform/mysql/retry_test.go

@@ -0,0 +1,143 @@
+package mysql
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/go-kit/log"
+	gmysql "github.com/go-sql-driver/mysql"
+	"github.com/jmoiron/sqlx"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// readOnlyErr returns a MySQL error that simulates a read-only database (error 1792).
+func readOnlyErr() error {
+	return &gmysql.MySQLError{Number: 1792, Message: "Cannot execute statement in a READ ONLY transaction."}
+}
+
+func TestTriggerFatalErrorCallsHandler(t *testing.T) {
+	var called atomic.Bool
+	var capturedErr atomic.Value
+	SetFatalErrorHandler(func(err error) {
+		called.Store(true)
+		capturedErr.Store(err)
+	})
+	t.Cleanup(func() { SetFatalErrorHandler(nil) })
+
+	testErr := errors.New("test read-only error")
+	TriggerFatalError(testErr)
+
+	assert.True(t, called.Load())
+	assert.Equal(t, testErr, capturedErr.Load())
+}
+
+func TestTriggerFatalErrorPanicsWithoutHandler(t *testing.T) {
+	SetFatalErrorHandler(nil)
+
+	assert.Panics(t, func() {
+		TriggerFatalError(errors.New("read-only"))
+	})
+}
+
+func TestTriggerFatalErrorFiresOnce(t *testing.T) {
+	var callCount atomic.Int32
+	SetFatalErrorHandler(func(_ error) {
+		callCount.Add(1)
+	})
+	t.Cleanup(func() { SetFatalErrorHandler(nil) })
+
+	var wg sync.WaitGroup
+	for range 100 {
+		wg.Go(func() {
+			TriggerFatalError(errors.New("read-only"))
+		})
+	}
+	wg.Wait()
+
+	assert.Equal(t, int32(1), callCount.Load())
+}
+
+func TestTransactionReadOnlyTriggersFatalError(t *testing.T) {
+	cases := []struct {
+		name      string
+		txFunc    func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error
+		setupMock func(mock sqlmock.Sqlmock)
+	}{
+		{
+			name: "WithRetryTxx read-only from fn",
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectBegin()
+				mock.ExpectRollback()
+			},
+			txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
+				return WithRetryTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
+					return readOnlyErr()
+				}, log.NewNopLogger())
+			},
+		},
+		{
+			name: "WithRetryTxx read-only from commit",
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectBegin()
+				mock.ExpectCommit().WillReturnError(readOnlyErr())
+			},
+			txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
+				return WithRetryTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
+					return nil
+				}, log.NewNopLogger())
+			},
+		},
+		{
+			name: "WithTxx read-only from fn",
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectBegin()
+				mock.ExpectRollback()
+			},
+			txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
+				return WithTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
+					return readOnlyErr()
+				}, log.NewNopLogger())
+			},
+		},
+		{
+			name: "WithTxx read-only from commit",
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectBegin()
+				mock.ExpectCommit().WillReturnError(readOnlyErr())
+			},
+			txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
+				return WithTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
+					return nil
+				}, log.NewNopLogger())
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var handlerCalled atomic.Bool
+			SetFatalErrorHandler(func(_ error) {
+				handlerCalled.Store(true)
+			})
+			t.Cleanup(func() { SetFatalErrorHandler(nil) })
+
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+			sqlxDB := sqlx.NewDb(db, "sqlmock")
+
+			tc.setupMock(mock)
+
+			err = tc.txFunc(t, sqlxDB, mock)
+
+			require.Error(t, err)
+			assert.True(t, IsReadOnlyError(err))
+			assert.True(t, handlerCalled.Load())
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}