fix: harden runtime cancellation, reloads, and decoding

cmd/coscout/commands/daemon.go

@@ -15,8 +15,10 @@
 package commands
 
 import (
+	"context"
 	"os"
 	"os/signal"
+	"sync"
 	"sync/atomic"
 	"syscall"
 
@@ -31,7 +33,56 @@ import (
 )
 
 type authState struct {
-	isAuthed atomic.Bool
+	lifecycle atomic.Int32
+	cancelMu  sync.Mutex
+	cancel    context.CancelFunc
+}
+
+const (
+	daemonIdle int32 = iota
+	daemonRunning
+	daemonStopping
+)
+
+func (s *authState) tryStart() (context.Context, bool) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	s.cancelMu.Lock()
+	defer s.cancelMu.Unlock()
+	if !s.lifecycle.CompareAndSwap(daemonIdle, daemonRunning) {
+		cancel()
+		return nil, false
+	}
+	s.cancel = cancel
+	return ctx, true
+}
+
+func (s *authState) requestStop() bool {
+	if !s.lifecycle.CompareAndSwap(daemonRunning, daemonStopping) {
+		return false
+	}
+
+	s.cancelMu.Lock()
+	cancel := s.cancel
+	s.cancelMu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	return true
+}
+
+func (s *authState) daemonStopped() {
+	s.cancelMu.Lock()
+	cancelRunAndMarkIdle(s.cancel, &s.lifecycle)
+	s.cancel = nil
+	s.cancelMu.Unlock()
+}
+
+func cancelRunAndMarkIdle(cancel context.CancelFunc, lifecycle *atomic.Int32) {
+	if cancel != nil {
+		cancel()
+	}
+	lifecycle.Store(daemonIdle)
 }
 
 func NewDaemonCommand(cfgPath *string) *cobra.Command {
@@ -42,7 +93,10 @@ func NewDaemonCommand(cfgPath *string) *cobra.Command {
 			storageDB := storage.NewBoltDB(config.GetDBPath())
 			confManager := config.InitConfManager(*cfgPath, &storageDB)
 
-			appConf := confManager.LoadOnce()
+			appConf, err := confManager.LoadStartup()
+			if err != nil {
+				log.Fatalf("Unable to load config file from %s: %v", *cfgPath, err)
+			}
 			log.Infof("Load config file from %s", *cfgPath)
 
 			registerChan := make(chan model.DeviceStatusResponse, 10)
@@ -66,28 +120,40 @@ func NewDaemonCommand(cfgPath *string) *cobra.Command {
 }
 
 func run(confManager *config.ConfManager, reqClient *api.RequestClient, registerChan chan model.DeviceStatusResponse) {
-	startChan := make(chan bool, 1)
-	exitChan := make(chan bool, 1)
 	errorChan := make(chan error, 100)
 
 	state := &authState{}
-	state.isAuthed.Store(false)
 	for deviceStatus := range registerChan {
 		if deviceStatus.Authorized {
-			log.Info("Device is authorized. Performing actions...")
+			startDaemon(state, confManager, reqClient, errorChan)
+			continue
+		}
 
-			if !state.isAuthed.Load() {
-				go daemon.Run(confManager, reqClient, startChan, exitChan, errorChan)
-				startChan <- true
-			}
-			state.isAuthed.Store(true)
-		} else {
-			log.Warn("Device is not authorized, waiting...")
+		stopDaemon(state)
+	}
+}
 
-			if state.isAuthed.Load() {
-				exitChan <- true
-			}
-			state.isAuthed.Store(false)
-		}
+func startDaemon(
+	state *authState,
+	confManager *config.ConfManager,
+	reqClient *api.RequestClient,
+	errorChan chan error,
+) {
+	log.Info("Device is authorized. Performing actions...")
+	ctx, started := state.tryStart()
+	if !started {
+		return
 	}
+
+	go func() {
+		defer state.daemonStopped()
+		if err := daemon.Run(ctx, confManager, reqClient, errorChan); err != nil {
+			log.Errorf("Daemon stopped: %v", err)
+		}
+	}()
+}
+
+func stopDaemon(state *authState) {
+	log.Warn("Device is not authorized, waiting...")
+	state.requestStop()
 }

cmd/coscout/commands/daemon_test.go

@@ -0,0 +1,93 @@
+// Copyright 2025 coScene
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package commands
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAuthStateStopFromFailedRunDoesNotCancelRestart(t *testing.T) {
+	t.Parallel()
+
+	state := &authState{}
+
+	firstCtx, started := state.tryStart()
+	require.True(t, started)
+	_, started = state.tryStart()
+	require.False(t, started)
+	require.True(t, state.requestStop())
+	require.False(t, state.requestStop())
+	require.ErrorIs(t, firstCtx.Err(), context.Canceled)
+	_, started = state.tryStart()
+	require.False(t, started)
+
+	state.daemonStopped()
+	secondCtx, started := state.tryStart()
+	require.True(t, started)
+	require.NoError(t, secondCtx.Err(), "a previous stop must not cancel a restarted daemon")
+}
+
+func TestAuthStateRepeatedAuthorizationTransitionsAreIdempotent(t *testing.T) {
+	t.Parallel()
+
+	state := &authState{}
+	for range 10 {
+		runCtx, started := state.tryStart()
+		require.True(t, started)
+
+		_, started = state.tryStart()
+		require.False(t, started, "duplicate authorization must not start another daemon")
+		require.True(t, state.requestStop())
+		require.False(t, state.requestStop(), "duplicate unauthorization must not request another stop")
+		require.ErrorIs(t, runCtx.Err(), context.Canceled)
+
+		state.daemonStopped()
+	}
+}
+
+func TestDaemonStoppedCancelsOldRunBeforeRestart(t *testing.T) {
+	t.Parallel()
+
+	state := &authState{}
+	firstCtx, started := state.tryStart()
+	require.True(t, started)
+
+	state.daemonStopped()
+
+	secondCtx, started := state.tryStart()
+	require.True(t, started)
+	require.ErrorIs(t, firstCtx.Err(), context.Canceled)
+	require.NoError(t, secondCtx.Err())
+}
+
+func TestCancelRunAndMarkIdleOrdersCancellationFirst(t *testing.T) {
+	t.Parallel()
+
+	var lifecycle atomic.Int32
+	lifecycle.Store(daemonRunning)
+	runCtx, cancel := context.WithCancel(t.Context())
+
+	cancelRunAndMarkIdle(func() {
+		require.Equal(t, daemonRunning, lifecycle.Load(), "idle became visible before the old run was canceled")
+		cancel()
+	}, &lifecycle)
+
+	require.ErrorIs(t, runCtx.Err(), context.Canceled)
+	require.Equal(t, daemonIdle, lifecycle.Load())
+}

internal/collector/collector.go

@@ -117,7 +117,14 @@ func Collect(ctx context.Context, reqClient *api.RequestClient, confManager *con
 						}
 					}()
 
-					appConfig := confManager.LoadWithRemote()
+					appConfig, configErr := confManager.LoadWithRemote()
+					if configErr != nil {
+						if appConfig == nil {
+							log.Errorf("Unable to load collector config: %v", configErr)
+							return
+						}
+						log.Warnf("Unable to reload collector config, using last-known-good config: %v", configErr)
+					}
 					getStorage := confManager.GetStorage()
 
 					err := handleRecordCaches(uploadChan, reqClient, appConfig, getStorage, recordSet)

internal/collector/upload.go

@@ -142,7 +142,14 @@ func Upload(ctx context.Context, reqClient *api.RequestClient, confManager *conf
 						return
 					}
 
-					appConfig := confManager.LoadWithRemote()
+					appConfig, configErr := confManager.LoadWithRemote()
+					if configErr != nil {
+						if appConfig == nil {
+							log.Errorf("Unable to load upload config: %v", configErr)
+							return
+						}
+						log.Warnf("Unable to reload upload config, using last-known-good config: %v", configErr)
+					}
 					if appConfig != nil {
 						enabled := appConfig.Upload.NetworkRule.Enabled
 						blackInterfaces := appConfig.Upload.NetworkRule.BlackInterfaces
@@ -219,7 +226,13 @@ func uploadFiles(ctx context.Context, reqClient *api.RequestClient, confManager
 	}
 
 	allCompleted := true
-	appConfig := confManager.LoadWithRemote()
+	appConfig, configErr := confManager.LoadWithRemote()
+	if configErr != nil {
+		if appConfig == nil {
+			return errors.Wrap(configErr, "load upload config")
+		}
+		log.Warnf("Unable to reload upload config, using last-known-good config: %v", configErr)
+	}
 	getStorage := confManager.GetStorage()
 
 	recordCache, err := recordCache.Reload()