From f1624c20a37ea76772ab26cd058f430b2c48b762 Mon Sep 17 00:00:00 2001 From: ryan Date: Mon, 9 Mar 2026 23:13:27 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E4=BB=A3=E7=90=86=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E5=8A=9F=E8=83=BD=EF=BC=8C=E5=8C=85=E6=8B=AC=E4=B8=BB?= =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=85=A5=E5=8F=A3=E3=80=81=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E5=8A=A0=E8=BD=BD=E3=80=81=E5=BF=83=E8=B7=B3=E6=A3=80=E6=B5=8B?= =?UTF-8?q?=E5=92=8C=E7=8A=B6=E6=80=81=E7=AE=A1=E7=90=86=EF=BC=8C=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0=E7=9B=B8=E5=85=B3=E6=9C=8D=E5=8A=A1=E5=92=8CAPI?= =?UTF-8?q?=E5=93=8D=E5=BA=94=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- atsf_agent/cmd/agent/main.go | 47 ++++++- atsf_agent/internal/agent/runner.go | 65 ++++++++++ atsf_agent/internal/config/config.go | 80 ++++++++++++ atsf_agent/internal/heartbeat/service.go | 28 ++++ atsf_agent/internal/httpclient/client.go | 97 ++++++++++++++ atsf_agent/internal/nginx/manager.go | 99 ++++++++++++++ atsf_agent/internal/protocol/agent_api.go | 6 + atsf_agent/internal/state/state.go | 96 ++++++++++++++ atsf_agent/internal/state/state_test.go | 21 +++ atsf_agent/internal/sync/service.go | 73 +++++++++++ atsf_agent/internal/sync/service_test.go | 150 ++++++++++++++++++++++ 11 files changed, 761 insertions(+), 1 deletion(-) create mode 100644 atsf_agent/internal/agent/runner.go create mode 100644 atsf_agent/internal/config/config.go create mode 100644 atsf_agent/internal/heartbeat/service.go create mode 100644 atsf_agent/internal/httpclient/client.go create mode 100644 atsf_agent/internal/nginx/manager.go create mode 100644 atsf_agent/internal/state/state.go create mode 100644 atsf_agent/internal/state/state_test.go create mode 100644 atsf_agent/internal/sync/service.go create mode 100644 atsf_agent/internal/sync/service_test.go diff --git a/atsf_agent/cmd/agent/main.go b/atsf_agent/cmd/agent/main.go index 38dd16da..ef038176 100644 --- a/atsf_agent/cmd/agent/main.go +++ b/atsf_agent/cmd/agent/main.go @@ -1,3 +1,48 @@ package main -func main() {} +import ( + "context" + "flag" + "log" + "os/signal" + "syscall" + + "atsflare-agent/internal/agent" + "atsflare-agent/internal/config" + "atsflare-agent/internal/heartbeat" + "atsflare-agent/internal/httpclient" + "atsflare-agent/internal/nginx" + "atsflare-agent/internal/state" + syncservice "atsflare-agent/internal/sync" +) + +func main() { + configPath := flag.String("config", "./agent.json", "agent config path") + flag.Parse() + + cfg, err := config.Load(*configPath) + if err != nil { + log.Fatal(err) + } + + client := httpclient.New(cfg.ServerURL, cfg.AgentToken, cfg.RequestTimeout) + stateStore := state.NewStore(cfg.StatePath) + runner := &agent.Runner{ + Config: cfg, + StateStore: stateStore, + HeartbeatService: heartbeat.New(client), + SyncService: syncservice.New(client, &nginx.Manager{ + RouteConfigPath: cfg.RouteConfigPath, + Executor: &nginx.ShellExecutor{ + Binary: cfg.NginxBinary, + }, + }, stateStore), + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + if err = runner.Run(ctx); err != nil && err != context.Canceled { + log.Fatal(err) + } +} diff --git a/atsf_agent/internal/agent/runner.go b/atsf_agent/internal/agent/runner.go new file mode 100644 index 00000000..93ef9128 --- /dev/null +++ b/atsf_agent/internal/agent/runner.go @@ -0,0 +1,65 @@ +package agent + +import ( + "context" + "time" + + "atsflare-agent/internal/config" + "atsflare-agent/internal/heartbeat" + "atsflare-agent/internal/protocol" + "atsflare-agent/internal/state" + syncservice "atsflare-agent/internal/sync" +) + +type Runner struct { + Config *config.Config + StateStore *state.Store + HeartbeatService *heartbeat.Service + SyncService *syncservice.Service +} + +func (r *Runner) Run(ctx context.Context) error { + nodeID, err := r.StateStore.EnsureNodeID() + if err != nil { + return err + } + if err = r.HeartbeatService.Register(ctx, r.nodePayload(nodeID)); err != nil { + return err + } + if err = r.SyncService.SyncOnce(ctx); err != nil { + return err + } + + heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval) + defer heartbeatTicker.Stop() + syncTicker := time.NewTicker(r.Config.SyncInterval) + defer syncTicker.Stop() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-heartbeatTicker.C: + if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil { + return err + } + case <-syncTicker.C: + if err = r.SyncService.SyncOnce(ctx); err != nil { + return err + } + } + } +} + +func (r *Runner) nodePayload(nodeID string) protocol.NodePayload { + snapshot, _ := r.StateStore.Load() + return protocol.NodePayload{ + NodeID: nodeID, + Name: r.Config.NodeName, + IP: r.Config.NodeIP, + AgentVersion: r.Config.AgentVersion, + NginxVersion: r.Config.NginxVersion, + CurrentVersion: snapshot.CurrentVersion, + LastError: snapshot.LastError, + } +} diff --git a/atsf_agent/internal/config/config.go b/atsf_agent/internal/config/config.go new file mode 100644 index 00000000..3eecd3ec --- /dev/null +++ b/atsf_agent/internal/config/config.go @@ -0,0 +1,80 @@ +package config + +import ( + "encoding/json" + "errors" + "os" + "path/filepath" + "time" +) + +type Config struct { + ServerURL string `json:"server_url"` + AgentToken string `json:"agent_token"` + NodeName string `json:"node_name"` + NodeIP string `json:"node_ip"` + AgentVersion string `json:"agent_version"` + NginxVersion string `json:"nginx_version"` + RouteConfigPath string `json:"route_config_path"` + StatePath string `json:"state_path"` + NginxBinary string `json:"nginx_binary"` + HeartbeatInterval time.Duration `json:"heartbeat_interval"` + SyncInterval time.Duration `json:"sync_interval"` + RequestTimeout time.Duration `json:"request_timeout"` +} + +func Load(path string) (*Config, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + cfg := &Config{} + if err = json.Unmarshal(data, cfg); err != nil { + return nil, err + } + applyDefaults(cfg) + if err = validate(cfg); err != nil { + return nil, err + } + return cfg, nil +} + +func applyDefaults(cfg *Config) { + if cfg.AgentVersion == "" { + cfg.AgentVersion = "dev" + } + if cfg.RouteConfigPath == "" { + cfg.RouteConfigPath = filepath.Clean("./atsflare_routes.conf") + } + if cfg.StatePath == "" { + cfg.StatePath = filepath.Clean("./atsf_agent_state.json") + } + if cfg.NginxBinary == "" { + cfg.NginxBinary = "nginx" + } + if cfg.HeartbeatInterval <= 0 { + cfg.HeartbeatInterval = 30 * time.Second + } + if cfg.SyncInterval <= 0 { + cfg.SyncInterval = 30 * time.Second + } + if cfg.RequestTimeout <= 0 { + cfg.RequestTimeout = 10 * time.Second + } +} + +func validate(cfg *Config) error { + if cfg.ServerURL == "" { + return errors.New("server_url 不能为空") + } + if cfg.AgentToken == "" { + return errors.New("agent_token 不能为空") + } + if cfg.NodeName == "" { + return errors.New("node_name 不能为空") + } + if cfg.NodeIP == "" { + return errors.New("node_ip 不能为空") + } + return nil +} diff --git a/atsf_agent/internal/heartbeat/service.go b/atsf_agent/internal/heartbeat/service.go new file mode 100644 index 00000000..18c87602 --- /dev/null +++ b/atsf_agent/internal/heartbeat/service.go @@ -0,0 +1,28 @@ +package heartbeat + +import ( + "context" + + "atsflare-agent/internal/protocol" +) + +type Client interface { + RegisterNode(ctx context.Context, payload protocol.NodePayload) error + Heartbeat(ctx context.Context, payload protocol.NodePayload) error +} + +type Service struct { + client Client +} + +func New(client Client) *Service { + return &Service{client: client} +} + +func (s *Service) Register(ctx context.Context, payload protocol.NodePayload) error { + return s.client.RegisterNode(ctx, payload) +} + +func (s *Service) Heartbeat(ctx context.Context, payload protocol.NodePayload) error { + return s.client.Heartbeat(ctx, payload) +} diff --git a/atsf_agent/internal/httpclient/client.go b/atsf_agent/internal/httpclient/client.go new file mode 100644 index 00000000..59e6927c --- /dev/null +++ b/atsf_agent/internal/httpclient/client.go @@ -0,0 +1,97 @@ +package httpclient + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "strings" + "time" + + "atsflare-agent/internal/protocol" +) + +type Client struct { + baseURL string + token string + httpClient *http.Client +} + +func New(baseURL string, token string, timeout time.Duration) *Client { + return &Client{ + baseURL: strings.TrimRight(baseURL, "/"), + token: token, + httpClient: &http.Client{ + Timeout: timeout, + }, + } +} + +func (c *Client) RegisterNode(ctx context.Context, payload protocol.NodePayload) error { + return c.postJSON(ctx, "/api/agent/nodes/register", payload, nil) +} + +func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) error { + return c.postJSON(ctx, "/api/agent/nodes/heartbeat", payload, nil) +} + +func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) { + resp := protocol.APIResponse[protocol.ActiveConfigResponse]{} + if err := c.getJSON(ctx, "/api/agent/config-versions/active", &resp); err != nil { + return nil, err + } + if !resp.Success { + return nil, errors.New(resp.Message) + } + return &resp.Data, nil +} + +func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error { + return c.postJSON(ctx, "/api/agent/apply-logs", payload, nil) +} + +func (c *Client) getJSON(ctx context.Context, path string, target any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil) + if err != nil { + return err + } + req.Header.Set("X-Agent-Token", c.token) + return c.do(req, target) +} + +func (c *Client) postJSON(ctx context.Context, path string, body any, target any) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Agent-Token", c.token) + return c.do(req, target) +} + +func (c *Client) do(req *http.Request, target any) error { + res, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return errors.New(res.Status) + } + if target == nil { + var wrapper protocol.APIResponse[json.RawMessage] + if err = json.NewDecoder(res.Body).Decode(&wrapper); err != nil { + return err + } + if !wrapper.Success { + return errors.New(wrapper.Message) + } + return nil + } + return json.NewDecoder(res.Body).Decode(target) +} diff --git a/atsf_agent/internal/nginx/manager.go b/atsf_agent/internal/nginx/manager.go new file mode 100644 index 00000000..c53254c0 --- /dev/null +++ b/atsf_agent/internal/nginx/manager.go @@ -0,0 +1,99 @@ +package nginx + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" +) + +type Executor interface { + Test(ctx context.Context) error + Reload(ctx context.Context) error +} + +type ShellExecutor struct { + Binary string +} + +func (e *ShellExecutor) Test(ctx context.Context) error { + cmd := exec.CommandContext(ctx, e.Binary, "-t") + output, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("nginx -t failed: %w: %s", err, string(output)) + } + return nil +} + +func (e *ShellExecutor) Reload(ctx context.Context) error { + cmd := exec.CommandContext(ctx, e.Binary, "-s", "reload") + output, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("nginx reload failed: %w: %s", err, string(output)) + } + return nil +} + +type Manager struct { + RouteConfigPath string + Executor Executor +} + +func (m *Manager) Apply(ctx context.Context, content string) error { + backupPath, hadExisting, err := m.backup() + if err != nil { + return err + } + if err = os.WriteFile(m.RouteConfigPath, []byte(content), 0o644); err != nil { + return err + } + if err = m.Executor.Test(ctx); err != nil { + _ = m.restore(backupPath, hadExisting) + return err + } + if err = m.Executor.Reload(ctx); err != nil { + _ = m.restore(backupPath, hadExisting) + return err + } + if backupPath != "" { + _ = os.Remove(backupPath) + } + return nil +} + +func (m *Manager) backup() (string, bool, error) { + if m.RouteConfigPath == "" { + return "", false, errors.New("route config path 不能为空") + } + if err := os.MkdirAll(filepath.Dir(m.RouteConfigPath), 0o755); err != nil { + return "", false, err + } + data, err := os.ReadFile(m.RouteConfigPath) + if err != nil { + if os.IsNotExist(err) { + return "", false, nil + } + return "", false, err + } + backupPath := m.RouteConfigPath + ".bak" + if err = os.WriteFile(backupPath, data, 0o644); err != nil { + return "", false, err + } + return backupPath, true, nil +} + +func (m *Manager) restore(backupPath string, hadExisting bool) error { + if hadExisting { + data, err := os.ReadFile(backupPath) + if err != nil { + return err + } + return os.WriteFile(m.RouteConfigPath, data, 0o644) + } + if err := os.Remove(m.RouteConfigPath); err != nil && !os.IsNotExist(err) { + return err + } + return nil +} diff --git a/atsf_agent/internal/protocol/agent_api.go b/atsf_agent/internal/protocol/agent_api.go index 5cc8ffa3..127f7c52 100644 --- a/atsf_agent/internal/protocol/agent_api.go +++ b/atsf_agent/internal/protocol/agent_api.go @@ -1,5 +1,11 @@ package protocol +type APIResponse[T any] struct { + Success bool `json:"success"` + Message string `json:"message"` + Data T `json:"data"` +} + type NodePayload struct { NodeID string `json:"node_id"` Name string `json:"name"` diff --git a/atsf_agent/internal/state/state.go b/atsf_agent/internal/state/state.go new file mode 100644 index 00000000..da078447 --- /dev/null +++ b/atsf_agent/internal/state/state.go @@ -0,0 +1,96 @@ +package state + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "sync" +) + +type Snapshot struct { + NodeID string `json:"node_id"` + CurrentVersion string `json:"current_version"` + CurrentChecksum string `json:"current_checksum"` + LastError string `json:"last_error"` +} + +type Store struct { + path string + mu sync.Mutex +} + +func NewStore(path string) *Store { + return &Store{path: filepath.Clean(path)} +} + +func (s *Store) Load() (*Snapshot, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.loadUnlocked() +} + +func (s *Store) EnsureNodeID() (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + + snapshot, err := s.loadUnlocked() + if err != nil { + return "", err + } + if snapshot.NodeID != "" { + return snapshot.NodeID, nil + } + snapshot.NodeID, err = newNodeID() + if err != nil { + return "", err + } + if err = s.saveUnlocked(snapshot); err != nil { + return "", err + } + return snapshot.NodeID, nil +} + +func (s *Store) Save(snapshot *Snapshot) error { + s.mu.Lock() + defer s.mu.Unlock() + return s.saveUnlocked(snapshot) +} + +func (s *Store) loadUnlocked() (*Snapshot, error) { + data, err := os.ReadFile(s.path) + if err != nil { + if os.IsNotExist(err) { + return &Snapshot{}, nil + } + return nil, err + } + snapshot := &Snapshot{} + if len(data) == 0 { + return snapshot, nil + } + if err = json.Unmarshal(data, snapshot); err != nil { + return nil, err + } + return snapshot, nil +} + +func (s *Store) saveUnlocked(snapshot *Snapshot) error { + if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil { + return err + } + data, err := json.MarshalIndent(snapshot, "", " ") + if err != nil { + return err + } + return os.WriteFile(s.path, data, 0o644) +} + +func newNodeID() (string, error) { + buf := make([]byte, 8) + if _, err := rand.Read(buf); err != nil { + return "", err + } + return "node-" + hex.EncodeToString(buf), nil +} diff --git a/atsf_agent/internal/state/state_test.go b/atsf_agent/internal/state/state_test.go new file mode 100644 index 00000000..def96f55 --- /dev/null +++ b/atsf_agent/internal/state/state_test.go @@ -0,0 +1,21 @@ +package state + +import ( + "path/filepath" + "testing" +) + +func TestEnsureNodeIDPersists(t *testing.T) { + store := NewStore(filepath.Join(t.TempDir(), "state.json")) + nodeID1, err := store.EnsureNodeID() + if err != nil { + t.Fatalf("EnsureNodeID failed: %v", err) + } + nodeID2, err := store.EnsureNodeID() + if err != nil { + t.Fatalf("EnsureNodeID second call failed: %v", err) + } + if nodeID1 == "" || nodeID1 != nodeID2 { + t.Fatal("expected node id to persist across calls") + } +} diff --git a/atsf_agent/internal/sync/service.go b/atsf_agent/internal/sync/service.go new file mode 100644 index 00000000..bf257627 --- /dev/null +++ b/atsf_agent/internal/sync/service.go @@ -0,0 +1,73 @@ +package sync + +import ( + "context" + + "atsflare-agent/internal/nginx" + "atsflare-agent/internal/protocol" + "atsflare-agent/internal/state" +) + +const ( + ApplyResultSuccess = "success" + ApplyResultFailed = "failed" +) + +type ConfigClient interface { + GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) + ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error +} + +type Service struct { + client ConfigClient + nginxManager *nginx.Manager + stateStore *state.Store +} + +func New(client ConfigClient, nginxManager *nginx.Manager, stateStore *state.Store) *Service { + return &Service{ + client: client, + nginxManager: nginxManager, + stateStore: stateStore, + } +} + +func (s *Service) SyncOnce(ctx context.Context) error { + snapshot, err := s.stateStore.Load() + if err != nil { + return err + } + config, err := s.client.GetActiveConfig(ctx) + if err != nil { + return err + } + if snapshot.CurrentVersion == config.Version && snapshot.CurrentChecksum == config.Checksum { + return nil + } + if err = s.nginxManager.Apply(ctx, config.RenderedConfig); err != nil { + snapshot.LastError = err.Error() + _ = s.stateStore.Save(snapshot) + reportErr := s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{ + NodeID: snapshot.NodeID, + Version: config.Version, + Result: ApplyResultFailed, + Message: err.Error(), + }) + if reportErr != nil { + return reportErr + } + return err + } + snapshot.CurrentVersion = config.Version + snapshot.CurrentChecksum = config.Checksum + snapshot.LastError = "" + if err = s.stateStore.Save(snapshot); err != nil { + return err + } + return s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{ + NodeID: snapshot.NodeID, + Version: config.Version, + Result: ApplyResultSuccess, + Message: "apply success", + }) +} diff --git a/atsf_agent/internal/sync/service_test.go b/atsf_agent/internal/sync/service_test.go new file mode 100644 index 00000000..cdeddf06 --- /dev/null +++ b/atsf_agent/internal/sync/service_test.go @@ -0,0 +1,150 @@ +package sync + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "atsflare-agent/internal/nginx" + "atsflare-agent/internal/protocol" + "atsflare-agent/internal/state" +) + +type fakeExecutor struct { + testErr error + reloadErr error +} + +type fakeClient struct { + config protocol.ActiveConfigResponse + reports []protocol.ApplyLogPayload +} + +func (f *fakeExecutor) Test(ctx context.Context) error { + return f.testErr +} + +func (f *fakeExecutor) Reload(ctx context.Context) error { + return f.reloadErr +} + +func (f *fakeClient) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) { + return &f.config, nil +} + +func (f *fakeClient) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error { + f.reports = append(f.reports, payload) + return nil +} + +func TestSyncOnceSuccess(t *testing.T) { + client := &fakeClient{ + config: protocol.ActiveConfigResponse{ + Version: "20260309-001", + Checksum: "checksum-1", + RenderedConfig: "server { listen 80; }", + CreatedAt: time.Now().Format(time.RFC3339), + }, + } + + stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json")) + nodeID, err := stateStore.EnsureNodeID() + if err != nil { + t.Fatalf("EnsureNodeID failed: %v", err) + } + snapshot, _ := stateStore.Load() + snapshot.NodeID = nodeID + if err = stateStore.Save(snapshot); err != nil { + t.Fatalf("failed to save initial state: %v", err) + } + + routePath := filepath.Join(t.TempDir(), "routes.conf") + service := New(client, &nginx.Manager{ + RouteConfigPath: routePath, + Executor: &fakeExecutor{}, + }, stateStore) + + if err = service.SyncOnce(context.Background()); err != nil { + t.Fatalf("SyncOnce failed: %v", err) + } + + data, err := os.ReadFile(routePath) + if err != nil { + t.Fatalf("failed to read route config: %v", err) + } + if string(data) != "server { listen 80; }" { + t.Fatal("expected rendered config to be written to route file") + } + snapshot, err = stateStore.Load() + if err != nil { + t.Fatalf("failed to load state: %v", err) + } + if snapshot.CurrentVersion != "20260309-001" || snapshot.CurrentChecksum != "checksum-1" { + t.Fatal("expected state store to persist current version and checksum") + } + if len(client.reports) != 1 || client.reports[0].Result != ApplyResultSuccess { + t.Fatal("expected successful apply report to be sent") + } +} + +func TestSyncOnceRollbackOnNginxFailure(t *testing.T) { + client := &fakeClient{ + config: protocol.ActiveConfigResponse{ + Version: "20260309-002", + Checksum: "checksum-2", + RenderedConfig: "server { listen 81; }", + CreatedAt: time.Now().Format(time.RFC3339), + }, + } + + tempDir := t.TempDir() + routePath := filepath.Join(tempDir, "routes.conf") + if err := os.WriteFile(routePath, []byte("server { listen 80; }"), 0o644); err != nil { + t.Fatalf("failed to seed route file: %v", err) + } + + stateStore := state.NewStore(filepath.Join(tempDir, "state.json")) + nodeID, err := stateStore.EnsureNodeID() + if err != nil { + t.Fatalf("EnsureNodeID failed: %v", err) + } + if err = stateStore.Save(&state.Snapshot{ + NodeID: nodeID, + CurrentVersion: "20260309-001", + CurrentChecksum: "checksum-1", + }); err != nil { + t.Fatalf("failed to seed state: %v", err) + } + + service := New(client, &nginx.Manager{ + RouteConfigPath: routePath, + Executor: &fakeExecutor{ + testErr: context.DeadlineExceeded, + }, + }, stateStore) + + err = service.SyncOnce(context.Background()) + if err == nil { + t.Fatal("expected SyncOnce to fail when nginx test fails") + } + + data, readErr := os.ReadFile(routePath) + if readErr != nil { + t.Fatalf("failed to read route file after rollback: %v", readErr) + } + if string(data) != "server { listen 80; }" { + t.Fatal("expected original route config to be restored after rollback") + } + snapshot, loadErr := stateStore.Load() + if loadErr != nil { + t.Fatalf("failed to load state: %v", loadErr) + } + if snapshot.CurrentVersion != "20260309-001" { + t.Fatal("expected failed sync not to overwrite current version") + } + if len(client.reports) != 1 || client.reports[0].Result != ApplyResultFailed { + t.Fatal("expected failed apply report to be sent") + } +}