diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md index 321189b..2d046fa 100644 --- a/.github/ISSUE_TEMPLATE/bug_report.md +++ b/.github/ISSUE_TEMPLATE/bug_report.md @@ -5,28 +5,44 @@ labels: bug --- ## Describe the bug -A clear and concise description of what the bug is. +What broke, and what were you trying to do? ## To Reproduce Steps to reproduce the behavior: -1. -2. -3. +1. +2. +3. ## Expected behavior What you expected to happen. ## Actual behavior -What actually happened (include errors, stack traces, and logs if possible). +What actually happened. Include errors, stack traces, and relevant logs if +possible. + +## Diagnostics +If the bundle starts, please run the experimental doctor command and attach the +terminal output plus the generated JSON report: + +```sh +docker exec any-sync-bundle any-sync-bundle doctor +``` + +The report is usually written to `./data/doctor/doctor_.json` on the +host. Please remove secrets before attaching configs or logs. ## Environment - Deployment: [AIO container | minimal container | binary] - + - Version/tag: -- OS/Arch: +- Compose file or start command: +- OS/Arch: +- Storage: [local Badger | S3/MinIO] ## Configuration - + ## Additional context Add any other context about the problem here. diff --git a/README.md b/README.md index 00053ce..d8eb629 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,7 @@ ```sh docker run -d \ + --name any-sync-bundle \ -e ANY_SYNC_BUNDLE_INIT_EXTERNAL_ADDRS="192.168.100.9" \ -p 33010:33010 \ -p 33020:33020/udp \ @@ -145,6 +146,7 @@ Edit `ANY_SYNC_BUNDLE_INIT_EXTERNAL_ADDRS` in the compose file before starting. | -------------------------- | ----------------------------------------- | ------- | | `./data/bundle-config.yml` | Service config + private keys | 🔴 Yes | | `./data/client-config.yml` | Client config (regenerated on each start) | 🟢 No | +| `./data/doctor/*.json` | Experimental diagnostic reports | 🟢 No | ### Storage Options @@ -221,6 +223,7 @@ All parameters available as binary flags or environment variables. See `./any-sy | ------------------ | ---------------------------------------------------------------- | | `start-bundle` | Start with external MongoDB/Redis | | `start-all-in-one` | Start with embedded MongoDB/Redis (used in all-in-one container) | +| `doctor` | Run experimental diagnostics against the already running bundle | ### Start Command Flags @@ -238,8 +241,39 @@ All parameters available as binary flags or environment variables. See `./any-sy | `--initial-s3-force-path-style` | Use path-style S3 URLs (required for MinIO)
‣ Default: `false`
‣ Environment Variable: `ANY_SYNC_BUNDLE_INIT_S3_FORCE_PATH_STYLE` | | `--initial-filenode-default-limit` | Storage limit per space in bytes
‣ Default: `1099511627776` (1 TiB)
‣ Environment Variable: `ANY_SYNC_BUNDLE_INIT_FILENODE_DEFAULT_LIMIT` | +### Doctor Command Flags + +| Flag | Description | +| ----------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `--bundle-config`, `-c` | Path to the bundle configuration YAML file, used to locate the doctor socket
‣ Default: `./data/bundle-config.yml`
‣ Environment Variable: `ANY_SYNC_BUNDLE_CONFIG` | + ## Operations +### Diagnostics + +Run diagnostics from inside the running container: + +```sh +docker exec any-sync-bundle any-sync-bundle doctor +``` + +The command connects to the running bundle over a local Unix socket next to +`bundle-config.yml`. In the default container layout this is `/data/bundle.sock`. + +```text +Connecting to running bundle + socket: /data/bundle.sock + status: connected + +Experimental: + doctor output and JSON report schema may change between releases. + +... + +[7/7] Report + written: /data/doctor/doctor_2026-05-22T14-33-10Z.json +``` + ### Backup & Recovery **Backup:** diff --git a/cmd/doctor.go b/cmd/doctor.go new file mode 100644 index 0000000..4635a4b --- /dev/null +++ b/cmd/doctor.go @@ -0,0 +1,35 @@ +package cmd + +import ( + "context" + "fmt" + "path/filepath" + + "github.com/urfave/cli/v2" + + "github.com/grishy/any-sync-bundle/doctor" +) + +func cmdDoctor(ctx context.Context) *cli.Command { + return &cli.Command{ + Name: "doctor", + Usage: "Run experimental diagnostics against the already running bundle process", + Description: "EXPERIMENTAL: command output and JSON report schema may change between releases.", + Flags: []cli.Flag{ + &cli.PathFlag{ + Name: flagStartBundleConfigPath, + Aliases: []string{"c"}, + Value: "./data/bundle-config.yml", + EnvVars: []string{"ANY_SYNC_BUNDLE_CONFIG"}, + Usage: "Path to the bundle configuration YAML file, used to locate the doctor socket", + }, + }, + Action: func(cCtx *cli.Context) error { + bundleConfigPath, err := filepath.Abs(cCtx.String(flagStartBundleConfigPath)) + if err != nil { + return fmt.Errorf("resolve bundle config path: %w", err) + } + return doctor.RunClient(ctx, doctor.SocketPath(bundleConfigPath), cCtx.App.Writer) + }, + } +} diff --git a/cmd/doctor_test.go b/cmd/doctor_test.go new file mode 100644 index 0000000..27efc60 --- /dev/null +++ b/cmd/doctor_test.go @@ -0,0 +1,90 @@ +package cmd + +import ( + "context" + "flag" + "path/filepath" + "testing" + + "github.com/urfave/cli/v2" +) + +func TestRootIncludesDoctorCommand(t *testing.T) { + app := Root(context.Background()) + + if app.Command("doctor") == nil { + t.Fatal("Root() does not include doctor command") + } +} + +func TestPrepareBundleConfigKeepsRuntimePaths(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + storagePath := filepath.Join(dir, "storage") + flagSet := flag.NewFlagSet("test", flag.ContinueOnError) + for _, cliFlag := range buildStartFlags() { + if err := cliFlag.Apply(flagSet); err != nil { + t.Fatalf("apply flag: %v", err) + } + } + if err := flagSet.Set(flagStartBundleConfigPath, bundleConfigPath); err != nil { + t.Fatalf("set bundle config path: %v", err) + } + if err := flagSet.Set(flagStartClientConfigPath, clientConfigPath); err != nil { + t.Fatalf("set client config path: %v", err) + } + if err := flagSet.Set(flagStartStoragePath, storagePath); err != nil { + t.Fatalf("set storage path: %v", err) + } + cCtx := cli.NewContext(cli.NewApp(), flagSet, nil) + + prepared, err := prepareBundleConfig(cCtx) + if err != nil { + t.Fatalf("prepareBundleConfig() error = %v", err) + } + + if prepared.Config == nil { + t.Fatal("prepared Config is nil") + } + if prepared.BundleConfigPath != bundleConfigPath { + t.Fatalf("bundle config path = %q, want %q", prepared.BundleConfigPath, bundleConfigPath) + } + if prepared.ClientConfigPath != clientConfigPath { + t.Fatalf("client config path = %q, want %q", prepared.ClientConfigPath, clientConfigPath) + } +} + +func TestPrepareBundleConfigStoresAbsoluteRuntimePaths(t *testing.T) { + dir := t.TempDir() + t.Chdir(dir) + flagSet := flag.NewFlagSet("test", flag.ContinueOnError) + for _, cliFlag := range buildStartFlags() { + if err := cliFlag.Apply(flagSet); err != nil { + t.Fatalf("apply flag: %v", err) + } + } + if err := flagSet.Set(flagStartBundleConfigPath, "data/bundle-config.yml"); err != nil { + t.Fatalf("set bundle config path: %v", err) + } + if err := flagSet.Set(flagStartClientConfigPath, "data/client-config.yml"); err != nil { + t.Fatalf("set client config path: %v", err) + } + cCtx := cli.NewContext(cli.NewApp(), flagSet, nil) + + prepared, err := prepareBundleConfig(cCtx) + if err != nil { + t.Fatalf("prepareBundleConfig() error = %v", err) + } + + wantBundleConfigPath := filepath.Join(dir, "data", "bundle-config.yml") + if prepared.BundleConfigPath != wantBundleConfigPath { + t.Fatalf("bundle config path = %q, want %q", + prepared.BundleConfigPath, wantBundleConfigPath) + } + wantClientConfigPath := filepath.Join(dir, "data", "client-config.yml") + if prepared.ClientConfigPath != wantClientConfigPath { + t.Fatalf("client config path = %q, want %q", + prepared.ClientConfigPath, wantClientConfigPath) + } +} diff --git a/cmd/root.go b/cmd/root.go index 53964cb..787f8fd 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -74,6 +74,7 @@ func Root(ctx context.Context) *cli.App { Commands: []*cli.Command{ cmdStartAllInOne(ctx), cmdStartBundle(ctx), + cmdDoctor(ctx), }, } } diff --git a/cmd/start.go b/cmd/start.go index 774580b..0f18ac8 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -23,6 +23,7 @@ import ( "go.uber.org/zap" bundleConfig "github.com/grishy/any-sync-bundle/config" + "github.com/grishy/any-sync-bundle/doctor" "github.com/grishy/any-sync-bundle/lightnode" ) @@ -31,6 +32,12 @@ type node struct { app *app.App } +type preparedBundleConfig struct { + Config *bundleConfig.Config + BundleConfigPath string + ClientConfigPath string +} + const ( serviceShutdownTimeout = 10 * time.Second clientConfigMode = 0o644 @@ -59,12 +66,12 @@ func cmdStartAllInOne(ctx context.Context) *cli.Command { printWelcomeMsg() - bundleCfg, err := prepareBundleConfig(cCtx) + preparedCfg, err := prepareBundleConfig(cCtx) if err != nil { return err } - applyAllInOneDefaults(bundleCfg) + applyAllInOneDefaults(preparedCfg.Config) // Start pprof server if enabled startPprofServer(ctx, cCtx) @@ -75,7 +82,7 @@ func cmdStartAllInOne(ctx context.Context) *cli.Command { } defer infra.stop() - return runBundleServices(ctx, bundleCfg) + return runBundleServices(ctx, preparedCfg) }, } } @@ -88,7 +95,7 @@ func cmdStartBundle(ctx context.Context) *cli.Command { Action: func(cCtx *cli.Context) error { printWelcomeMsg() - bundleCfg, err := prepareBundleConfig(cCtx) + preparedCfg, err := prepareBundleConfig(cCtx) if err != nil { return err } @@ -96,12 +103,13 @@ func cmdStartBundle(ctx context.Context) *cli.Command { // Start pprof server if enabled startPprofServer(ctx, cCtx) - return runBundleServices(ctx, bundleCfg) + return runBundleServices(ctx, preparedCfg) }, } } -func runBundleServices(ctx context.Context, bundleCfg *bundleConfig.Config) error { +func runBundleServices(ctx context.Context, preparedCfg *preparedBundleConfig) error { + bundleCfg := preparedCfg.Config printConfigurationInfo(bundleCfg) cfgNodes := bundleCfg.NodeConfigs() @@ -118,11 +126,41 @@ func runBundleServices(ctx context.Context, bundleCfg *bundleConfig.Config) erro return err } + doctorRunner, err := doctor.NewLiveRuntimeRunner(doctor.LiveRuntimeConfig{ + BundleConfig: bundleCfg, + BundleConfigPath: preparedCfg.BundleConfigPath, + ClientConfigPath: preparedCfg.ClientConfigPath, + Build: doctor.BuildInfo{ + Version: version, + Commit: commit, + Date: date, + }, + FileNode: bundle.FileNode, + }) + if err != nil { + shutdownServices(apps) + return fmt.Errorf("create doctor runner: %w", err) + } + doctorServer := doctor.NewServer(doctor.ServerConfig{ + SocketPath: doctor.SocketPath(preparedCfg.BundleConfigPath), + Runner: doctorRunner, + }) + if startErr := doctorServer.Start(ctx); startErr != nil { + shutdownServices(apps) + return fmt.Errorf("start doctor server: %w", startErr) + } + emitBundleEvent(bundleReadyEvent) printStartupMsg() <-ctx.Done() + doctorCtx, doctorCancel := context.WithTimeout(context.Background(), serviceShutdownTimeout) + if closeErr := doctorServer.Close(doctorCtx); closeErr != nil { + log.Warn("doctor server shutdown failed", zap.Error(closeErr)) + } + doctorCancel() + shutdownServices(apps) emitBundleEvent(bundleShutdownCompleteEvent) printShutdownMsg() @@ -131,19 +169,30 @@ func runBundleServices(ctx context.Context, bundleCfg *bundleConfig.Config) erro return nil } -func prepareBundleConfig(cCtx *cli.Context) (*bundleConfig.Config, error) { - bundleCfg := loadOrCreateConfig(cCtx, log) - clientCfgPath := cCtx.String(flagStartClientConfigPath) +func prepareBundleConfig(cCtx *cli.Context) (*preparedBundleConfig, error) { + bundleCfgPath, err := filepath.Abs(cCtx.String(flagStartBundleConfigPath)) + if err != nil { + return nil, fmt.Errorf("resolve bundle config path: %w", err) + } + clientCfgPath, err := filepath.Abs(cCtx.String(flagStartClientConfigPath)) + if err != nil { + return nil, fmt.Errorf("resolve client config path: %w", err) + } + + bundleCfg := loadOrCreateConfig(cCtx, log, bundleCfgPath) - if err := writeClientConfig(bundleCfg, clientCfgPath); err != nil { - return nil, err + if writeErr := writeClientConfig(bundleCfg, clientCfgPath); writeErr != nil { + return nil, writeErr } - return bundleCfg, nil + return &preparedBundleConfig{ + Config: bundleCfg, + BundleConfigPath: bundleCfgPath, + ClientConfigPath: clientCfgPath, + }, nil } -func loadOrCreateConfig(cCtx *cli.Context, log logger.CtxLogger) *bundleConfig.Config { - cfgPath := cCtx.String(flagStartBundleConfigPath) +func loadOrCreateConfig(cCtx *cli.Context, log logger.CtxLogger, cfgPath string) *bundleConfig.Config { log.Info("loading config") if _, err := os.Stat(cfgPath); err == nil { diff --git a/compose.aio.yml b/compose.aio.yml index 849f791..17d79a9 100644 --- a/compose.aio.yml +++ b/compose.aio.yml @@ -2,6 +2,7 @@ # # Usage: # docker compose -f compose.aio.yml up -d +# docker exec any-sync-bundle-aio any-sync-bundle doctor # # The bundle image already contains MongoDB and Redis. Only the bundle service is required. diff --git a/compose.external.yml b/compose.external.yml index c6325ee..72bdae0 100644 --- a/compose.external.yml +++ b/compose.external.yml @@ -2,6 +2,7 @@ # # Usage: # docker compose -f compose.external.yml up -d +# docker exec any-sync-bundle any-sync-bundle doctor # # The bundle container runs alongside dedicated MongoDB and Redis instances. diff --git a/compose.s3.yml b/compose.s3.yml index 94a2481..4146db1 100644 --- a/compose.s3.yml +++ b/compose.s3.yml @@ -2,6 +2,7 @@ # # Usage: # docker compose -f compose.s3.yml up -d +# docker exec any-sync-bundle-aio any-sync-bundle doctor # # This runs the all-in-one bundle (embedded MongoDB/Redis) with MinIO for S3 storage. # MinIO Console: http://localhost:9001 (minioadmin/minioadmin) diff --git a/compose.traefik.yml b/compose.traefik.yml index ac8dad1..5356ee9 100644 --- a/compose.traefik.yml +++ b/compose.traefik.yml @@ -6,6 +6,7 @@ # 1. Edit ANY_SYNC_BUNDLE_INIT_EXTERNAL_ADDRS below (replace with your server's IP/hostname) # 2. Run: docker compose -f compose.traefik.yml up -d # 3. Get client config: cat ./data/client-config.yml +# 4. Run diagnostics: docker exec any-sync-bundle-aio any-sync-bundle doctor # # Debugging: # - View Traefik dashboard: http://localhost:8080 diff --git a/config/bundle.go b/config/bundle.go index 65de033..e1244be 100644 --- a/config/bundle.go +++ b/config/bundle.go @@ -30,6 +30,11 @@ const ( // oneTiB is one tebibyte (2^40 bytes), used as the default filenode storage limit. oneTiB = 1024 * 1024 * 1024 * 1024 + + defaultListenTCPAddr = "0.0.0.0:33010" + defaultListenUDPAddr = "0.0.0.0:33020" + defaultCoordinatorMongoDBName = "coordinator" + defaultConsensusMongoDBName = "consensus" ) type Config struct { @@ -334,16 +339,16 @@ func newBundleConfig(cfg *CreateOptions) *Config { StoragePath: cfg.StorePath, Account: newAcc(netKey), Network: NetworkConfig{ - ListenTCPAddr: "0.0.0.0:33010", - ListenUDPAddr: "0.0.0.0:33020", + ListenTCPAddr: defaultListenTCPAddr, + ListenUDPAddr: defaultListenUDPAddr, }, Coordinator: CoordinatorConfig{ MongoConnect: cfg.MongoURI, - MongoDatabase: "coordinator", + MongoDatabase: defaultCoordinatorMongoDBName, }, Consensus: ConsensusConfig{ MongoConnect: mongoConsensusURI.String(), - MongoDatabase: "consensus", + MongoDatabase: defaultConsensusMongoDBName, }, FileNode: FileNodeConfig{ RedisConnect: cfg.RedisURI, diff --git a/doctor/client.go b/doctor/client.go new file mode 100644 index 0000000..d6bb571 --- /dev/null +++ b/doctor/client.go @@ -0,0 +1,53 @@ +package doctor + +import ( + "context" + "fmt" + "io" + "net" + "net/http" +) + +func unixHTTPClient(socketPath string) *http.Client { + transport := &http.Transport{ + DialContext: func(ctx context.Context, _ string, _ string) (net.Conn, error) { + var dialer net.Dialer + return dialer.DialContext(ctx, "unix", socketPath) + }, + DisableCompression: true, + } + return &http.Client{Transport: transport} +} + +func RunClient(ctx context.Context, socketPath string, out io.Writer) error { + fmt.Fprintln(out, "Connecting to running bundle") + fmt.Fprintf(out, " socket: %s\n", socketPath) + + request, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://doctor/doctor/run", nil) + if err != nil { + return fmt.Errorf("create doctor request: %w", err) + } + + client := unixHTTPClient(socketPath) + defer client.CloseIdleConnections() + + response, err := client.Do(request) + if err != nil { + return fmt.Errorf("connect to doctor socket %q: %w", socketPath, err) + } + defer response.Body.Close() + + if response.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + return fmt.Errorf("doctor request failed: %s: %s", response.Status, string(body)) + } + fmt.Fprintln(out, " status: connected") + fmt.Fprintln(out) + + _, copyErr := io.Copy(out, response.Body) + if copyErr != nil { + return fmt.Errorf("read doctor stream: %w", copyErr) + } + + return nil +} diff --git a/doctor/index.go b/doctor/index.go new file mode 100644 index 0000000..f0fbea3 --- /dev/null +++ b/doctor/index.go @@ -0,0 +1,664 @@ +package doctor + +import ( + "fmt" + "sort" + "strconv" + "strings" + + "github.com/anyproto/any-sync-filenode/index/indexproto" +) + +const ( + redisIndexInfoField = "info" + redisIndexGroupPrefix = "g:" + redisIndexSpacePrefix = "s:" + redisIndexCIDPrefix = "c:" + redisIndexFilePrefix = "f:" +) + +type IndexSnapshot struct { + Hashes map[string]map[string]string + Values map[string]string +} + +type CIDRefs map[string]uint64 + +type cidEntryState struct { + entry *indexproto.CidEntry + problem Problem + hasProblem bool + missing bool +} + +type spaceInspection struct { + space SpaceReport + files []FileReport + expectedRefs CIDRefs + uniqueCIDs map[string]struct{} + uniqueCIDBytes map[string]uint64 + problems []Problem +} + +//nolint:cyclop,funlen,gocognit,gocyclo,nestif // The Redis index scan is one coherent ownership/ref/counter pass. +func InspectIndexSnapshot(snapshot IndexSnapshot) (Inventory, []Problem, error) { + inventory := Inventory{} + problems := []Problem{} + groupsByID := map[string]*GroupReport{} + groupInfosByID := map[string]*indexproto.GroupEntry{} + groupRefsByID := map[string]CIDRefs{} + groupExpectedRefsByID := map[string]CIDRefs{} + groupCIDBytesByID := map[string]map[string]uint64{} + spaceGroupByID := map[string]string{} + listedSpaceIDsByGroupID := map[string]map[string]struct{}{} + seenSpaceIDs := map[string]struct{}{} + cidEntryCache := map[string]cidEntryState{} + cidProblemReported := map[string]struct{}{} + globalSpacesByCID := map[string]map[string]struct{}{} + + hashKeys := sortedMapKeys(snapshot.Hashes) + for _, key := range hashKeys { + if !strings.HasPrefix(key, redisIndexGroupPrefix) { + continue + } + groupID := parseIndexHashID(key, redisIndexGroupPrefix) + report := &GroupReport{ + ID: groupID, + Status: StatusOK, + } + fields := snapshot.Hashes[key] + if encodedInfo, ok := fields[redisIndexInfoField]; ok { + entry := &indexproto.GroupEntry{} + if err := entry.UnmarshalVT([]byte(encodedInfo)); err != nil { + report.Status = StatusProblem + report.IndexProblems++ + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: fmt.Sprintf("group info cannot be decoded: %v", err), + }) + } else { + report.ID = entry.GetGroupId() + if report.ID == "" { + report.ID = groupID + } + report.Limit = entry.GetLimit() + report.AccountLimit = entry.GetAccountLimit() + groupInfosByID[report.ID] = entry + listedSpaces := listedSpaceIDsByGroupID[report.ID] + if listedSpaces == nil { + listedSpaces = map[string]struct{}{} + listedSpaceIDsByGroupID[report.ID] = listedSpaces + } + for _, spaceID := range entry.GetSpaceIds() { + listedSpaces[spaceID] = struct{}{} + if existingGroupID, exists := spaceGroupByID[spaceID]; exists && existingGroupID != report.ID { + report.Status = StatusProblem + report.IndexProblems++ + problems = append(problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space %s is listed in multiple groups: %s and %s", + spaceID, existingGroupID, report.ID), + }) + } else { + spaceGroupByID[spaceID] = report.ID + } + } + } + } else { + report.Status = StatusProblem + report.IndexProblems++ + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: "group info is missing", + }) + } + refs, refProblems := cidRefsFromFields("group", report.ID, fields) + if len(refProblems) > 0 { + report.Status = StatusProblem + report.IndexProblems += uint64(len(refProblems)) + problems = append(problems, refProblems...) + } + groupRefsByID[report.ID] = refs + groupsByID[report.ID] = report + } + + for _, key := range sortedMapKeys(snapshot.Values) { + if !strings.HasPrefix(key, redisIndexCIDPrefix) { + continue + } + cid := strings.TrimPrefix(key, redisIndexCIDPrefix) + loadCIDEntry(cid, snapshot.Values, cidEntryCache) + } + + for _, key := range hashKeys { + if !strings.HasPrefix(key, redisIndexSpacePrefix) { + continue + } + inspection := inspectSpaceHash( + key, + snapshot.Hashes[key], + snapshot.Values, + spaceGroupByID, + cidEntryCache, + cidProblemReported, + ) + problems = append(problems, inspection.problems...) + seenSpaceIDs[inspection.space.ID] = struct{}{} + inventory.Spaces = append(inventory.Spaces, inspection.space) + inventory.Files = append(inventory.Files, inspection.files...) + + for cid := range inspection.uniqueCIDs { + spaces := globalSpacesByCID[cid] + if spaces == nil { + spaces = map[string]struct{}{} + globalSpacesByCID[cid] = spaces + } + spaces[inspection.space.ID] = struct{}{} + } + + if inspection.space.GroupID == "" { + continue + } + group := groupsByID[inspection.space.GroupID] + if group == nil { + group = &GroupReport{ID: inspection.space.GroupID, Status: StatusOK} + groupsByID[inspection.space.GroupID] = group + } + group.Spaces++ + group.Files += inspection.space.Files + expectedRefs := groupExpectedRefsByID[inspection.space.GroupID] + if expectedRefs == nil { + expectedRefs = CIDRefs{} + groupExpectedRefsByID[inspection.space.GroupID] = expectedRefs + } + for cid, ref := range inspection.expectedRefs { + expectedRefs[cid] += ref + } + cidBytes := groupCIDBytesByID[inspection.space.GroupID] + if cidBytes == nil { + cidBytes = map[string]uint64{} + groupCIDBytesByID[inspection.space.GroupID] = cidBytes + } + for cid, size := range inspection.uniqueCIDBytes { + cidBytes[cid] = size + } + if inspection.space.Status == StatusProblem { + group.Status = StatusProblem + group.IndexProblems += inspection.space.IndexProblems + } + } + + for groupID, listedSpaces := range listedSpaceIDsByGroupID { + group := groupsByID[groupID] + if group == nil { + continue + } + for _, spaceID := range sortedMapKeys(listedSpaces) { + if _, ok := seenSpaceIDs[spaceID]; ok { + continue + } + group.Status = StatusProblem + group.IndexProblems++ + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: fmt.Sprintf("space %s is listed in group %s but has no space entry", spaceID, groupID), + }) + } + } + + for groupID, group := range groupsByID { + expectedRefs := groupExpectedRefsByID[groupID] + group.CIDs = uint64(len(expectedRefs)) + group.Bytes = sumCIDBytes(groupCIDBytesByID[groupID]) + groupProblems := compareCIDRefs("group", groupID, expectedRefs, groupRefsByID[groupID]) + groupProblems = append(groupProblems, compareGroupCounters(groupID, groupInfosByID[groupID], group)...) + if len(groupProblems) > 0 { + group.Status = StatusProblem + group.IndexProblems += uint64(len(groupProblems)) + problems = append(problems, groupProblems...) + } + } + + globalProblems := inspectGlobalCIDRefs(cidEntryCache, globalSpacesByCID, cidProblemReported) + problems = append(problems, globalProblems...) + + groupIDs := sortedMapKeys(groupsByID) + for _, groupID := range groupIDs { + group := *groupsByID[groupID] + if group.Files == 0 && group.CIDs == 0 && group.Status == StatusOK { + group.Status = StatusEmpty + } + inventory.Groups = append(inventory.Groups, group) + } + + sort.Slice(inventory.Spaces, func(i, j int) bool { + return inventory.Spaces[i].ID < inventory.Spaces[j].ID + }) + sort.Slice(inventory.Files, func(i, j int) bool { + if inventory.Files[i].SpaceID == inventory.Files[j].SpaceID { + return inventory.Files[i].ID < inventory.Files[j].ID + } + return inventory.Files[i].SpaceID < inventory.Files[j].SpaceID + }) + + return inventory, problems, nil +} + +//nolint:funlen,gocognit,nestif // Space inspection mirrors one Redis hash and keeps derived values together. +func inspectSpaceHash( + key string, + fields map[string]string, + values map[string]string, + spaceGroupByID map[string]string, + cidEntryCache map[string]cidEntryState, + cidProblemReported map[string]struct{}, +) spaceInspection { + spaceID := parseIndexHashID(key, redisIndexSpacePrefix) + inspection := spaceInspection{ + space: SpaceReport{ + ID: spaceID, + Status: StatusOK, + }, + expectedRefs: CIDRefs{}, + uniqueCIDs: map[string]struct{}{}, + uniqueCIDBytes: map[string]uint64{}, + } + declaredFileCount := uint64(0) + declaredCIDCount := uint64(0) + declaredBytes := uint64(0) + hasSpaceInfo := false + + if encodedInfo, ok := fields[redisIndexInfoField]; ok { + entry := &indexproto.SpaceEntry{} + if err := entry.UnmarshalVT([]byte(encodedInfo)); err != nil { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space info cannot be decoded: %v", err), + }) + } else { + hasSpaceInfo = true + inspection.space.GroupID = entry.GetGroupId() + inspection.space.Limit = entry.GetLimit() + declaredFileCount = uint64(entry.GetFileCount()) + declaredCIDCount = entry.GetCidCount() + declaredBytes = entry.GetSize() + } + } else { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: "space info is missing", + }) + } + + listedGroupID, listedInGroup := spaceGroupByID[spaceID] + if inspection.space.GroupID == "" { + if listedInGroup { + inspection.space.GroupID = listedGroupID + } + } else { + if listedInGroup { + if listedGroupID != inspection.space.GroupID { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space %s group mismatch: spaceInfo=%s groupEntry=%s", + spaceID, inspection.space.GroupID, listedGroupID), + }) + } + } else { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space %s belongs to group %s but no group lists it", + spaceID, inspection.space.GroupID), + }) + } + } + if inspection.space.GroupID == "" { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space %s has no group assignment", spaceID), + }) + } + + spaceRefs, refProblems := cidRefsFromFields("space", spaceID, fields) + for _, problem := range refProblems { + markSpaceProblem(&inspection.space, &inspection.problems, problem) + } + + fieldKeys := sortedMapKeys(fields) + for _, field := range fieldKeys { + if !strings.HasPrefix(field, redisIndexFilePrefix) { + continue + } + fileID := strings.TrimPrefix(field, redisIndexFilePrefix) + entry := &indexproto.FileEntry{} + if err := entry.UnmarshalVT([]byte(fields[field])); err != nil { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeFile, + ID: fileID, + Issue: fmt.Sprintf("file entry in space %s cannot be decoded: %v", spaceID, err), + }) + continue + } + + file := FileReport{ + ID: fileID, + SpaceID: spaceID, + GroupID: inspection.space.GroupID, + Size: entry.GetSize(), + CIDs: append([]string(nil), entry.GetCids()...), + CIDSizes: map[string]uint64{}, + } + fileSize := uint64(0) + fileSizeKnown := true + for _, cid := range file.CIDs { + inspection.expectedRefs[cid]++ + inspection.uniqueCIDs[cid] = struct{}{} + state := loadCIDEntry(cid, values, cidEntryCache) + if state.hasProblem { + if _, ok := cidProblemReported[cid]; !ok { + markSpaceProblem(&inspection.space, &inspection.problems, state.problem) + cidProblemReported[cid] = struct{}{} + } else { + inspection.space.Status = StatusProblem + inspection.space.IndexProblems++ + } + if state.missing { + file.MissingCIDIndex = append(file.MissingCIDIndex, cid) + inspection.space.MissingCIDIndex++ + } + fileSizeKnown = false + continue + } + fileSize += state.entry.GetSize() + file.CIDSizes[cid] = state.entry.GetSize() + inspection.uniqueCIDBytes[cid] = state.entry.GetSize() + } + if fileSizeKnown { + file.Size = fileSize + if entry.GetSize() != fileSize { + markSpaceProblem(&inspection.space, &inspection.problems, Problem{ + Scope: problemScopeFile, + ID: fileID, + Issue: fmt.Sprintf("file size mismatch in space %s file %s: index=%d actual=%d", + spaceID, fileID, entry.GetSize(), fileSize), + }) + } + } + inspection.files = append(inspection.files, file) + } + + inspection.space.Files = uint64(len(inspection.files)) + inspection.space.CIDs = uint64(len(inspection.uniqueCIDs)) + inspection.space.Bytes = sumCIDBytes(inspection.uniqueCIDBytes) + + for _, problem := range compareCIDRefs("space", spaceID, inspection.expectedRefs, spaceRefs) { + markSpaceProblem(&inspection.space, &inspection.problems, problem) + } + if hasSpaceInfo { + for _, problem := range compareSpaceCounters( + spaceID, + declaredFileCount, + declaredCIDCount, + declaredBytes, + inspection.space, + ) { + markSpaceProblem(&inspection.space, &inspection.problems, problem) + } + } + if inspection.space.Files == 0 && inspection.space.CIDs == 0 && inspection.space.Status == StatusOK { + inspection.space.Status = StatusEmpty + } + + return inspection +} + +func compareCIDRefs(scope string, id string, expected CIDRefs, actual CIDRefs) []Problem { + problems := []Problem{} + for _, cid := range sortedMapKeys(expected) { + expectedRef := expected[cid] + actualRef, ok := actual[cid] + if !ok { + problems = append(problems, Problem{ + Scope: scope, + ID: id, + Issue: fmt.Sprintf("%s is missing CID ref %s", scope, cid), + }) + continue + } + if actualRef != expectedRef { + problems = append(problems, Problem{ + Scope: scope, + ID: id, + Issue: fmt.Sprintf("%s CID ref mismatch for %s: index=%d actual=%d", + scope, cid, actualRef, expectedRef), + }) + } + } + for _, cid := range sortedMapKeys(actual) { + if _, ok := expected[cid]; ok { + continue + } + problems = append(problems, Problem{ + Scope: scope, + ID: id, + Issue: fmt.Sprintf("%s has stale CID ref %s", scope, cid), + }) + } + return problems +} + +func compareGroupCounters(groupID string, info *indexproto.GroupEntry, group *GroupReport) []Problem { + problems := []Problem{} + if info != nil { + if info.GetCidCount() != group.CIDs { + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: fmt.Sprintf("group CID count mismatch: index=%d actual=%d", info.GetCidCount(), group.CIDs), + }) + } + if info.GetSize() != group.Bytes { + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: fmt.Sprintf("group byte count mismatch: index=%d actual=%d", info.GetSize(), group.Bytes), + }) + } + } + if group.Limit > 0 && group.Bytes > group.Limit { + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: fmt.Sprintf("group byte size exceeds limit: size=%d limit=%d", group.Bytes, group.Limit), + }) + } + if group.AccountLimit > 0 && group.Bytes > group.AccountLimit { + problems = append(problems, Problem{ + Scope: problemScopeGroup, + ID: groupID, + Issue: fmt.Sprintf("group byte size exceeds account limit: size=%d accountLimit=%d", + group.Bytes, group.AccountLimit), + }) + } + return problems +} + +func compareSpaceCounters( + spaceID string, + declaredFileCount uint64, + declaredCIDCount uint64, + declaredBytes uint64, + space SpaceReport, +) []Problem { + problems := []Problem{} + if declaredFileCount != space.Files { + problems = append(problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space file count mismatch: index=%d actual=%d", declaredFileCount, space.Files), + }) + } + if declaredCIDCount != space.CIDs { + problems = append(problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space CID count mismatch: index=%d actual=%d", declaredCIDCount, space.CIDs), + }) + } + if declaredBytes != space.Bytes { + problems = append(problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space byte count mismatch: index=%d actual=%d", declaredBytes, space.Bytes), + }) + } + if space.Limit > 0 && space.Bytes > space.Limit { + problems = append(problems, Problem{ + Scope: problemScopeSpace, + ID: spaceID, + Issue: fmt.Sprintf("space byte size exceeds limit: size=%d limit=%d", space.Bytes, space.Limit), + }) + } + return problems +} + +func inspectGlobalCIDRefs( + cidEntryCache map[string]cidEntryState, + globalSpacesByCID map[string]map[string]struct{}, + cidProblemReported map[string]struct{}, +) []Problem { + problems := []Problem{} + for _, cid := range sortedMapKeys(cidEntryCache) { + state := cidEntryCache[cid] + if state.hasProblem { + if _, ok := cidProblemReported[cid]; !ok { + problems = append(problems, state.problem) + } + continue + } + if len(globalSpacesByCID[cid]) == 0 { + problems = append(problems, Problem{ + Scope: problemScopeCID, + ID: cid, + Issue: fmt.Sprintf("global CID entry %s is not referenced by any file", cid), + }) + continue + } + expectedRefs := int64(len(globalSpacesByCID[cid])) + if int64(state.entry.GetRefs()) != expectedRefs { + problems = append(problems, Problem{ + Scope: problemScopeCID, + ID: cid, + Issue: fmt.Sprintf("global CID refs mismatch: index=%d actual=%d", + state.entry.GetRefs(), expectedRefs), + }) + } + } + return problems +} + +func loadCIDEntry(cid string, values map[string]string, cache map[string]cidEntryState) cidEntryState { + if state, ok := cache[cid]; ok { + return state + } + encoded, ok := values[redisIndexCIDKey(cid)] + if !ok { + state := cidEntryState{ + hasProblem: true, + missing: true, + problem: Problem{ + Scope: problemScopeCID, + ID: cid, + Issue: fmt.Sprintf("file references CID %s but global index entry is missing", cid), + }, + } + cache[cid] = state + return state + } + entry := &indexproto.CidEntry{} + if err := entry.UnmarshalVT([]byte(encoded)); err != nil { + state := cidEntryState{ + hasProblem: true, + problem: Problem{ + Scope: problemScopeCID, + ID: cid, + Issue: fmt.Sprintf("global CID entry %s cannot be decoded: %v", cid, err), + }, + } + cache[cid] = state + return state + } + state := cidEntryState{entry: entry} + cache[cid] = state + return state +} + +func cidRefsFromFields(scope string, id string, fields map[string]string) (CIDRefs, []Problem) { + cids := CIDRefs{} + problems := []Problem{} + for field, value := range fields { + if !strings.HasPrefix(field, redisIndexCIDPrefix) { + continue + } + cid := strings.TrimPrefix(field, redisIndexCIDPrefix) + ref, err := strconv.ParseUint(value, 10, 64) + if err != nil { + problems = append(problems, Problem{ + Scope: scope, + ID: id, + Issue: fmt.Sprintf("%s CID ref %s cannot be decoded: %v", scope, cid, err), + }) + continue + } + cids[cid] = ref + } + return cids, problems +} + +func redisIndexCIDKey(cid string) string { + return redisIndexCIDPrefix + cid +} + +func redisIndexScanPattern(prefix string) string { + return prefix + "*" +} + +func markSpaceProblem(space *SpaceReport, problems *[]Problem, problem Problem) { + space.Status = StatusProblem + space.IndexProblems++ + *problems = append(*problems, problem) +} + +func sumCIDBytes(cids map[string]uint64) uint64 { + var bytes uint64 + for _, size := range cids { + bytes += size + } + return bytes +} + +func parseIndexHashID(key string, prefix string) string { + value := strings.TrimPrefix(key, prefix) + if idx := strings.Index(value, ".{"); idx >= 0 { + return value[:idx] + } + return value +} + +func sortedMapKeys[V any](values map[string]V) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} diff --git a/doctor/index_test.go b/doctor/index_test.go new file mode 100644 index 0000000..233bbea --- /dev/null +++ b/doctor/index_test.go @@ -0,0 +1,711 @@ +package doctor + +import ( + "strings" + "testing" + + "github.com/anyproto/any-sync-filenode/index" + "github.com/anyproto/any-sync-filenode/index/indexproto" +) + +func TestInspectIndexSnapshotBuildsInventoryAndFindsMissingCIDIndex(t *testing.T) { + groupID := "account1" + spaceID := "space1" + fileID := "file1" + cidOK := "bafy-ok" + cidMissing := "bafy-missing" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}}, + ), + redisIndexCIDKey(cidOK): "1", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 2, Size: 10}, + ), + index.FileKey(fileID): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidOK, cidMissing}, Size: 10}, + ), + redisIndexCIDKey(cidOK): "1", + redisIndexCIDKey(cidMissing): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidOK): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 5, Refs: 1}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + if len(inventory.Groups) != 1 { + t.Fatalf("groups = %d, want 1", len(inventory.Groups)) + } + if inventory.Groups[0].ID != groupID { + t.Fatalf("group ID = %q, want %q", inventory.Groups[0].ID, groupID) + } + if inventory.Groups[0].Spaces != 1 { + t.Fatalf("group spaces = %d, want 1", inventory.Groups[0].Spaces) + } + if inventory.Groups[0].Files != 1 { + t.Fatalf("group files = %d, want 1", inventory.Groups[0].Files) + } + if inventory.Groups[0].CIDs != 2 { + t.Fatalf("group cids = %d, want 2", inventory.Groups[0].CIDs) + } + if inventory.Groups[0].Bytes != 5 { + t.Fatalf("group bytes = %d, want 5", inventory.Groups[0].Bytes) + } + if len(inventory.Spaces) != 1 { + t.Fatalf("spaces = %d, want 1", len(inventory.Spaces)) + } + + space := inventory.Spaces[0] + if space.ID != spaceID { + t.Fatalf("space ID = %q, want %q", space.ID, spaceID) + } + if space.GroupID != groupID { + t.Fatalf("space group = %q, want %q", space.GroupID, groupID) + } + if space.Files != 1 { + t.Fatalf("space files = %d, want 1", space.Files) + } + if space.CIDs != 2 { + t.Fatalf("space cids = %d, want 2", space.CIDs) + } + if space.MissingCIDIndex != 1 { + t.Fatalf("missing CID index = %d, want 1", space.MissingCIDIndex) + } + if space.Status != StatusProblem { + t.Fatalf("space status = %q, want %q", space.Status, StatusProblem) + } + + if len(problems) == 0 { + t.Fatal("problems = 0, want at least one missing CID problem") + } + if !strings.Contains(problems[0].Issue, cidMissing) { + t.Fatalf("problem issue = %q, want missing CID", problems[0].Issue) + } +} + +func TestInspectIndexSnapshotKeepsEmptySpacesVisible(t *testing.T) { + groupID := "account1" + spaceID := "empty-space" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}}, + ), + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry(t, &indexproto.SpaceEntry{GroupId: groupID}), + }, + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + if len(problems) != 0 { + t.Fatalf("problems = %d, want 0", len(problems)) + } + if len(inventory.Spaces) != 1 { + t.Fatalf("spaces = %d, want 1", len(inventory.Spaces)) + } + if inventory.Spaces[0].Status != StatusEmpty { + t.Fatalf("space status = %q, want %q", inventory.Spaces[0].Status, StatusEmpty) + } +} + +func TestInspectIndexSnapshotFindsCIDRefAndCounterProblems(t *testing.T) { + groupID := "account1" + spaceID := "space1" + fileID := "file1" + cidOK := "bafy-ok" + cidStale := "bafy-stale" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}, CidCount: 2, Size: 20}, + ), + redisIndexCIDKey(cidStale): "1", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 2, CidCount: 2, Size: 20}, + ), + index.FileKey(fileID): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidOK}, Size: 10}, + ), + redisIndexCIDKey(cidStale): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidOK): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + if len(inventory.Spaces) != 1 { + t.Fatalf("spaces = %d, want 1", len(inventory.Spaces)) + } + space := inventory.Spaces[0] + if space.Status != StatusProblem { + t.Fatalf("space status = %q, want %q", space.Status, StatusProblem) + } + if space.IndexProblems < 5 { + t.Fatalf("space index problems = %d, want at least 5", space.IndexProblems) + } + + wantIssues := []string{ + "space is missing CID ref " + cidOK, + "space has stale CID ref " + cidStale, + "group is missing CID ref " + cidOK, + "group has stale CID ref " + cidStale, + "space file count mismatch", + "space CID count mismatch", + "space byte count mismatch", + "group CID count mismatch", + "group byte count mismatch", + } + for _, want := range wantIssues { + if !problemIssuesContain(problems, want) { + t.Fatalf("problems do not contain %q:\n%v", want, problems) + } + } +} + +func TestInspectIndexSnapshotDeduplicatesGroupCIDCounts(t *testing.T) { + groupID := "account1" + spaceIDOne := "space1" + spaceIDTwo := "space2" + cidShared := "bafy-shared" + keyOne := index.Key{GroupId: groupID, SpaceId: spaceIDOne} + keyTwo := index.Key{GroupId: groupID, SpaceId: spaceIDTwo} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(keyOne): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{ + GroupId: groupID, + SpaceIds: []string{spaceIDOne, spaceIDTwo}, + CidCount: 1, + Size: 10, + }, + ), + redisIndexCIDKey(cidShared): "2", + }, + index.SpaceKey(keyOne): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidShared}, Size: 10}, + ), + redisIndexCIDKey(cidShared): "1", + }, + index.SpaceKey(keyTwo): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey("file2"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidShared}, Size: 10}, + ), + redisIndexCIDKey(cidShared): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidShared): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 2}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + if len(problems) != 0 { + t.Fatalf("problems = %d, want 0: %v", len(problems), problems) + } + if len(inventory.Groups) != 1 { + t.Fatalf("groups = %d, want 1", len(inventory.Groups)) + } + if inventory.Groups[0].CIDs != 1 { + t.Fatalf("group cids = %d, want 1", inventory.Groups[0].CIDs) + } + if inventory.Groups[0].Files != 2 { + t.Fatalf("group files = %d, want 2", inventory.Groups[0].Files) + } +} + +func TestInspectIndexSnapshotFindsCIDRefCountMismatches(t *testing.T) { + groupID := "account1" + spaceID := "space1" + cidShared := "bafy-shared" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}, CidCount: 1, Size: 10}, + ), + redisIndexCIDKey(cidShared): "1", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 2, CidCount: 1, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidShared}, Size: 10}, + ), + index.FileKey("file2"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidShared}, Size: 10}, + ), + redisIndexCIDKey(cidShared): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidShared): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + } + + _, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + for _, want := range []string{ + "space CID ref mismatch for " + cidShared + ": index=1 actual=2", + "group CID ref mismatch for " + cidShared + ": index=1 actual=2", + } { + if !problemIssuesContain(problems, want) { + t.Fatalf("problems do not contain %q:\n%v", want, problems) + } + } +} + +func TestInspectIndexSnapshotComputesBytesFromCIDEntries(t *testing.T) { + groupID := "account1" + spaceID := "space1" + cidShared := "bafy-shared" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}, CidCount: 1, Size: 10}, + ), + redisIndexCIDKey(cidShared): "2", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 2, CidCount: 1, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidShared}, Size: 10}, + ), + index.FileKey("file2"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidShared}, Size: 10}, + ), + redisIndexCIDKey(cidShared): "2", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidShared): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + if len(problems) != 0 { + t.Fatalf("problems = %d, want 0: %v", len(problems), problems) + } + if inventory.Spaces[0].Bytes != 10 { + t.Fatalf("space bytes = %d, want 10", inventory.Spaces[0].Bytes) + } + if inventory.Groups[0].Bytes != 10 { + t.Fatalf("group bytes = %d, want 10", inventory.Groups[0].Bytes) + } +} + +func TestInspectIndexSnapshotFindsCorruptGlobalCIDEntry(t *testing.T) { + groupID := "account1" + spaceID := "space1" + cidCorrupt := "bafy-corrupt" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}, CidCount: 1, Size: 10}, + ), + redisIndexCIDKey(cidCorrupt): "1", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidCorrupt}, Size: 10}, + ), + redisIndexCIDKey(cidCorrupt): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidCorrupt): "not-protobuf", + }, + } + + _, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + if !problemIssuesContain(problems, "global CID entry "+cidCorrupt+" cannot be decoded") { + t.Fatalf("problems do not include corrupt global CID entry: %v", problems) + } +} + +func TestInspectIndexSnapshotFindsFileSizeMismatchFromCIDEntries(t *testing.T) { + groupID := "account1" + spaceID := "space1" + cidOne := "bafy-one" + cidTwo := "bafy-two" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}, CidCount: 2, Size: 10}, + ), + redisIndexCIDKey(cidOne): "1", + redisIndexCIDKey(cidTwo): "1", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 2, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidOne, cidTwo}, Size: 99}, + ), + redisIndexCIDKey(cidOne): "1", + redisIndexCIDKey(cidTwo): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidOne): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 4, Refs: 1}), + redisIndexCIDKey(cidTwo): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 6, Refs: 1}), + }, + } + + _, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + if !problemIssuesContain(problems, "file size mismatch in space space1 file file1: index=99 actual=10") { + t.Fatalf("problems do not include file size mismatch: %v", problems) + } +} + +func TestInspectIndexSnapshotAssignsSpaceGroupFromGroupEntrySpaceIDs(t *testing.T) { + groupIDOne := "account1" + groupIDTwo := "account2" + spaceID := "space1" + cidOne := "bafy-one" + keyOne := index.Key{GroupId: groupIDOne, SpaceId: spaceID} + keyTwo := index.Key{GroupId: groupIDTwo, SpaceId: "other-space"} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(keyOne): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupIDOne, SpaceIds: []string{spaceID}, CidCount: 1, Size: 10}, + ), + redisIndexCIDKey(cidOne): "1", + }, + index.GroupKey(keyTwo): { + redisIndexInfoField: mustMarshalGroupEntry(t, &indexproto.GroupEntry{GroupId: groupIDTwo}), + }, + index.SpaceKey(keyOne): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidOne}, Size: 10}, + ), + redisIndexCIDKey(cidOne): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidOne): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + if len(problems) != 0 { + t.Fatalf("problems = %d, want 0: %v", len(problems), problems) + } + if inventory.Spaces[0].GroupID != groupIDOne { + t.Fatalf("space group = %q, want %q", inventory.Spaces[0].GroupID, groupIDOne) + } +} + +func TestInspectIndexSnapshotReportsUnassignedSpace(t *testing.T) { + groupID := "account1" + spaceID := "space1" + cidOne := "bafy-one" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{"other-space"}}, + ), + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidOne}, Size: 10}, + ), + redisIndexCIDKey(cidOne): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidOne): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + if inventory.Spaces[0].GroupID != "" { + t.Fatalf("space group = %q, want empty", inventory.Spaces[0].GroupID) + } + if !problemIssuesContain(problems, "space "+spaceID+" has no group assignment") { + t.Fatalf("problems do not include unassigned space: %v", problems) + } +} + +func TestInspectIndexSnapshotReportsRequiredInfoAndOwnershipProblems(t *testing.T) { + groupID := "account1" + otherGroupID := "account2" + spaceID := "space1" + missingSpaceID := "missing-space" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + otherKey := index.Key{GroupId: otherGroupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry(t, &indexproto.GroupEntry{ + GroupId: groupID, + SpaceIds: []string{missingSpaceID}, + }), + }, + index.GroupKey(otherKey): {}, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry(t, &indexproto.SpaceEntry{ + GroupId: otherGroupID, + }), + }, + }, + } + + _, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + for _, want := range []string{ + "group info is missing", + "space " + missingSpaceID + " is listed in group " + groupID + " but has no space entry", + "space " + spaceID + " belongs to group " + otherGroupID + " but no group lists it", + } { + if !problemIssuesContain(problems, want) { + t.Fatalf("problems do not contain %q:\n%v", want, problems) + } + } +} + +func TestInspectIndexSnapshotReportsGlobalCIDEntriesWithoutFileRefs(t *testing.T) { + cidOrphan := "bafy-orphan" + cidCorrupt := "bafy-corrupt-orphan" + snapshot := IndexSnapshot{ + Values: map[string]string{ + redisIndexCIDKey(cidOrphan): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + redisIndexCIDKey(cidCorrupt): "not-protobuf", + }, + } + + _, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + for _, want := range []string{ + "global CID entry " + cidCorrupt + " cannot be decoded", + "global CID entry " + cidOrphan + " is not referenced by any file", + } { + if !problemIssuesContain(problems, want) { + t.Fatalf("problems do not contain %q:\n%v", want, problems) + } + } +} + +func TestInspectIndexSnapshotReportsLimitProblems(t *testing.T) { + groupID := "account1" + spaceID := "space1" + cidOne := "bafy-one" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry(t, &indexproto.GroupEntry{ + GroupId: groupID, + SpaceIds: []string{spaceID}, + CidCount: 1, + Size: 10, + Limit: 5, + AccountLimit: 7, + }), + redisIndexCIDKey(cidOne): "1", + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry(t, &indexproto.SpaceEntry{ + GroupId: groupID, + FileCount: 1, + CidCount: 1, + Size: 10, + Limit: 5, + }), + index.FileKey("file1"): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidOne}, Size: 10}, + ), + redisIndexCIDKey(cidOne): "1", + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidOne): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + } + + inventory, problems, err := InspectIndexSnapshot(snapshot) + if err != nil { + t.Fatalf("InspectIndexSnapshot() error = %v", err) + } + + if inventory.Groups[0].Limit != 5 { + t.Fatalf("group limit = %d, want 5", inventory.Groups[0].Limit) + } + if inventory.Spaces[0].Limit != 5 { + t.Fatalf("space limit = %d, want 5", inventory.Spaces[0].Limit) + } + for _, want := range []string{ + "group byte size exceeds limit: size=10 limit=5", + "group byte size exceeds account limit: size=10 accountLimit=7", + "space byte size exceeds limit: size=10 limit=5", + } { + if !problemIssuesContain(problems, want) { + t.Fatalf("problems do not contain %q:\n%v", want, problems) + } + } +} + +func mustMarshalGroupEntry(t *testing.T, entry *indexproto.GroupEntry) string { + t.Helper() + data, err := entry.MarshalVT() + if err != nil { + t.Fatalf("marshal group entry: %v", err) + } + return string(data) +} + +func mustMarshalSpaceEntry(t *testing.T, entry *indexproto.SpaceEntry) string { + t.Helper() + data, err := entry.MarshalVT() + if err != nil { + t.Fatalf("marshal space entry: %v", err) + } + return string(data) +} + +func mustMarshalFileEntry(t *testing.T, entry *indexproto.FileEntry) string { + t.Helper() + data, err := entry.MarshalVT() + if err != nil { + t.Fatalf("marshal file entry: %v", err) + } + return string(data) +} + +func mustMarshalCIDEntry(t *testing.T, entry *indexproto.CidEntry) string { + t.Helper() + data, err := entry.MarshalVT() + if err != nil { + t.Fatalf("marshal CID entry: %v", err) + } + return string(data) +} + +func problemIssuesContain(problems []Problem, want string) bool { + for _, problem := range problems { + if strings.Contains(problem.Issue, want) { + return true + } + } + return false +} diff --git a/doctor/live.go b/doctor/live.go new file mode 100644 index 0000000..e9c5544 --- /dev/null +++ b/doctor/live.go @@ -0,0 +1,412 @@ +package doctor + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/anyproto/any-sync-filenode/redisprovider" + filenodestore "github.com/anyproto/any-sync-filenode/store" + "github.com/anyproto/any-sync/app" + "github.com/anyproto/any-sync/commonfile/fileblockstore" + blocks "github.com/ipfs/go-block-format" + "github.com/ipfs/go-cid" + "github.com/redis/go-redis/v9" + "go.mongodb.org/mongo-driver/mongo" + "go.mongodb.org/mongo-driver/mongo/options" + "go.mongodb.org/mongo-driver/mongo/readpref" + + bundleconfig "github.com/grishy/any-sync-bundle/config" +) + +type LiveRuntimeConfig struct { + BundleConfig *bundleconfig.Config + BundleConfigPath string + ClientConfigPath string + Build BuildInfo + FileNode *app.App +} + +func NewLiveRuntimeRunner(cfg LiveRuntimeConfig) (*RuntimeRunner, error) { + redisClient, err := redisClientFromFileNode(cfg.FileNode) + if err != nil { + return nil, err + } + blockStore, err := blockStoreFromFileNode(cfg.FileNode) + if err != nil { + return nil, err + } + + return NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg.BundleConfig, + BundleConfigPath: cfg.BundleConfigPath, + ClientConfigPath: cfg.ClientConfigPath, + Build: cfg.Build, + Now: func() time.Time { return time.Now().UTC() }, + LoadIndexSnapshot: func(ctx context.Context) (IndexSnapshot, error) { + return LoadIndexSnapshotFromRedis(ctx, redisClient) + }, + ProbeBlocks: func(ctx context.Context, inventory Inventory) (BlockProbeResult, error) { + return ProbeBlocksWithStore(ctx, inventory, blockStore) + }, + ProbeRuntime: func(ctx context.Context) RuntimeProbeResult { + return ProbeRuntime(ctx, cfg.BundleConfig, redisClient) + }, + }), nil +} + +func LoadIndexSnapshotFromRedis(ctx context.Context, client redis.UniversalClient) (IndexSnapshot, error) { + if client == nil { + return IndexSnapshot{}, errors.New("redis client is required") + } + + snapshot := IndexSnapshot{ + Hashes: map[string]map[string]string{}, + Values: map[string]string{}, + } + for _, pattern := range []string{ + redisIndexScanPattern(redisIndexGroupPrefix), + redisIndexScanPattern(redisIndexSpacePrefix), + } { + keys, err := scanRedisKeys(ctx, client, pattern) + if err != nil { + return IndexSnapshot{}, err + } + for _, key := range keys { + fields, fieldsErr := client.HGetAll(ctx, key).Result() + if fieldsErr != nil { + return IndexSnapshot{}, fmt.Errorf("read redis hash %s: %w", key, fieldsErr) + } + snapshot.Hashes[key] = fields + } + } + + keys, err := scanRedisKeys(ctx, client, redisIndexScanPattern(redisIndexCIDPrefix)) + if err != nil { + return IndexSnapshot{}, err + } + for _, key := range keys { + value, valueErr := client.Get(ctx, key).Result() + if valueErr != nil { + if errors.Is(valueErr, redis.Nil) { + continue + } + return IndexSnapshot{}, fmt.Errorf("read redis value %s: %w", key, valueErr) + } + snapshot.Values[key] = value + } + + return snapshot, nil +} + +func ProbeBlocksWithStore( + ctx context.Context, + inventory Inventory, + blockStore filenodestore.Store, +) (BlockProbeResult, error) { + if blockStore == nil { + return BlockProbeResult{}, errors.New("filenode block store is required") + } + + result := BlockProbeResult{ + MissingByFile: map[string][]string{}, + CorruptByFile: map[string][]string{}, + } + for _, file := range inventory.Files { + for _, cidString := range file.CIDs { + if err := ctx.Err(); err != nil { + return result, err + } + decodedCID, err := cid.Decode(cidString) + if err != nil { + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("CID %s cannot be decoded: %v", cidString, err), + }) + continue + } + + blockCtx := fileblockstore.CtxWithSpaceId(ctx, file.SpaceID) + blockCtx = fileblockstore.CtxWithFileId(blockCtx, file.ID) + block, err := blockStore.Get(blockCtx, decodedCID) + result.Checked++ + if err == nil { + if blockProblems := inspectBlock(file, cidString, decodedCID, block); len(blockProblems) > 0 { + key := fileReportKey(file.SpaceID, file.ID) + result.CorruptByFile[key] = append(result.CorruptByFile[key], cidString) + result.Problems = append(result.Problems, blockProblems...) + } + continue + } + if errors.Is(err, fileblockstore.ErrCIDNotFound) { + key := fileReportKey(file.SpaceID, file.ID) + result.MissingByFile[key] = append(result.MissingByFile[key], cidString) + continue + } + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("block %s cannot be read: %v", cidString, err), + }) + } + } + return result, nil +} + +func inspectBlock(file FileReport, cidString string, expectedCID cid.Cid, block blocks.Block) []Problem { + if block == nil { + return []Problem{{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("block %s returned an empty block", cidString), + }} + } + + problems := []Problem{} + if !block.Cid().Equals(expectedCID) { + problems = append(problems, Problem{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("block %s returned CID %s", cidString, block.Cid().String()), + }) + } + + data := block.RawData() + actualCID, err := expectedCID.Prefix().Sum(data) + if err != nil { + problems = append(problems, Problem{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("block %s content cannot be hashed: %v", cidString, err), + }) + } else if !actualCID.Equals(expectedCID) { + problems = append(problems, Problem{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("block %s content does not match its CID", cidString), + }) + } + + expectedSize, ok := file.CIDSizes[cidString] + if !ok { + if len(file.CIDs) == 1 { + expectedSize = file.Size + ok = expectedSize > 0 + } + } + if ok { + actualSize := uint64(len(data)) + if actualSize != expectedSize { + problems = append(problems, Problem{ + Scope: problemScopeFile, + ID: file.ID, + Issue: fmt.Sprintf("block %s size mismatch: index=%d actual=%d", + cidString, expectedSize, actualSize), + }) + } + } + + return problems +} + +func ProbeRuntime(ctx context.Context, cfg *bundleconfig.Config, redisClient redis.UniversalClient) RuntimeProbeResult { + result := RuntimeProbeResult{ + Mongo: StatusOK, + Redis: StatusOK, + RedisBloom: StatusOK, + Storage: StatusOK, + } + if cfg == nil { + result.Mongo = StatusProblem + result.Redis = StatusProblem + result.RedisBloom = StatusProblem + result.Storage = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: "bundle config is not available for runtime probes", + }) + return result + } + if storagePathErr := checkStoragePath(cfg.StoragePath); storagePathErr != nil { + result.Storage = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: fmt.Sprintf("storage path check failed: %v", storagePathErr), + }) + } else if storageLayoutErr := checkStorageLayout(cfg); storageLayoutErr != nil { + result.Storage = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: fmt.Sprintf("storage layout check failed: %v", storageLayoutErr), + }) + } + + if err := pingMongo(ctx, cfg.Coordinator.MongoConnect); err != nil { + result.Mongo = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: fmt.Sprintf("coordinator mongo ping failed: %v", err), + }) + } + if cfg.Consensus.MongoConnect != cfg.Coordinator.MongoConnect { + if err := pingMongo(ctx, cfg.Consensus.MongoConnect); err != nil { + result.Mongo = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: fmt.Sprintf("consensus mongo ping failed: %v", err), + }) + } + } + if redisClient == nil { + result.Redis = StatusProblem + result.RedisBloom = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: "redis client is not available", + }) + return result + } + if err := redisClient.Ping(ctx).Err(); err != nil { + result.Redis = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: fmt.Sprintf("redis ping failed: %v", err), + }) + } + if err := redisClient.Do(ctx, "BF.EXISTS", "_doctor_bloom_probe", "probe"). + Err(); err != nil && + !errors.Is(err, redis.Nil) { + result.RedisBloom = StatusProblem + result.Problems = append(result.Problems, Problem{ + Scope: problemScopeRuntime, + Issue: fmt.Sprintf("redis bloom probe failed: %v", err), + }) + } + return result +} + +func checkStoragePath(path string) error { + info, err := os.Stat(path) + if err != nil { + return err + } + if !info.IsDir() { + return fmt.Errorf("%s is not a directory", path) + } + + tmp, err := os.CreateTemp(path, ".doctor-write-test-*.tmp") + if err != nil { + return err + } + tmpPath := tmp.Name() + if closeErr := tmp.Close(); closeErr != nil { + _ = os.Remove(tmpPath) + return closeErr + } + if removeErr := os.Remove(tmpPath); removeErr != nil { + return fmt.Errorf("remove write test file %s: %w", filepath.Base(tmpPath), removeErr) + } + return nil +} + +func checkStorageLayout(cfg *bundleconfig.Config) error { + requiredDirs := []string{ + filepath.Join(cfg.StoragePath, "network-store", "coordinator"), + filepath.Join(cfg.StoragePath, "network-store", "consensus"), + filepath.Join(cfg.StoragePath, "network-store", "filenode"), + filepath.Join(cfg.StoragePath, "network-store", "sync"), + filepath.Join(cfg.StoragePath, "storage-sync"), + } + if cfg.FileNode.S3 == nil { + requiredDirs = append(requiredDirs, filepath.Join(cfg.StoragePath, "storage-file")) + } + + for _, path := range requiredDirs { + if err := checkReadableDirectory(path); err != nil { + rel, relErr := filepath.Rel(cfg.StoragePath, path) + if relErr != nil { + rel = path + } + return fmt.Errorf("%s: %w", rel, err) + } + } + return nil +} + +func checkReadableDirectory(path string) error { + info, err := os.Stat(path) + if err != nil { + return err + } + if !info.IsDir() { + return errors.New("not a directory") + } + if _, readErr := os.ReadDir(path); readErr != nil { + return readErr + } + return nil +} + +func scanRedisKeys(ctx context.Context, client redis.UniversalClient, pattern string) ([]string, error) { + var cursor uint64 + keys := []string{} + for { + if err := ctx.Err(); err != nil { + return nil, err + } + batch, nextCursor, err := client.Scan(ctx, cursor, pattern, 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan redis keys %s: %w", pattern, err) + } + keys = append(keys, batch...) + cursor = nextCursor + if cursor == 0 { + return keys, nil + } + } +} + +func redisClientFromFileNode(fileNode *app.App) (redis.UniversalClient, error) { + if fileNode == nil { + return nil, errors.New("filenode app is not available") + } + component := fileNode.Component(redisprovider.CName) + provider, ok := component.(redisprovider.RedisProvider) + if !ok { + return nil, errors.New("filenode redis provider component is not available") + } + return provider.Redis(), nil +} + +func blockStoreFromFileNode(fileNode *app.App) (filenodestore.Store, error) { + if fileNode == nil { + return nil, errors.New("filenode app is not available") + } + component := fileNode.Component(fileblockstore.CName) + blockStore, ok := component.(filenodestore.Store) + if !ok { + return nil, errors.New("filenode block store component is not available") + } + return blockStore, nil +} + +func pingMongo(ctx context.Context, uri string) error { + pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + client, err := mongo.Connect(pingCtx, options.Client().ApplyURI(uri)) + if err != nil { + return err + } + defer func() { + disconnectCtx, disconnectCancel := context.WithTimeout(context.Background(), 3*time.Second) + defer disconnectCancel() + _ = client.Disconnect(disconnectCtx) + }() + + return client.Ping(pingCtx, readpref.Primary()) +} diff --git a/doctor/live_test.go b/doctor/live_test.go new file mode 100644 index 0000000..0f8313e --- /dev/null +++ b/doctor/live_test.go @@ -0,0 +1,142 @@ +package doctor + +import ( + "context" + "path/filepath" + "testing" + + filenodestore "github.com/anyproto/any-sync-filenode/store" + "github.com/anyproto/any-sync/app" + "github.com/anyproto/any-sync/commonfile/fileblockstore" + blocks "github.com/ipfs/go-block-format" + "github.com/ipfs/go-cid" + + bundleconfig "github.com/grishy/any-sync-bundle/config" +) + +func TestProbeRuntimeReportsMissingStoragePath(t *testing.T) { + cfg := &bundleconfig.Config{ + StoragePath: filepath.Join(t.TempDir(), "missing-storage"), + Coordinator: bundleconfig.CoordinatorConfig{ + MongoConnect: "://bad-mongo-uri", + }, + Consensus: bundleconfig.ConsensusConfig{ + MongoConnect: "://bad-mongo-uri", + }, + } + + result := ProbeRuntime(context.Background(), cfg, nil) + + if result.Storage != StatusProblem { + t.Fatalf("storage status = %q, want %q", result.Storage, StatusProblem) + } + if !problemIssuesContain(result.Problems, "storage path check failed") { + t.Fatalf("problems do not include storage failure: %v", result.Problems) + } +} + +func TestProbeRuntimeReportsMissingStorageLayout(t *testing.T) { + storagePath := t.TempDir() + cfg := &bundleconfig.Config{ + StoragePath: storagePath, + Coordinator: bundleconfig.CoordinatorConfig{ + MongoConnect: "://bad-mongo-uri", + }, + Consensus: bundleconfig.ConsensusConfig{ + MongoConnect: "://bad-mongo-uri", + }, + } + + result := ProbeRuntime(context.Background(), cfg, nil) + + if result.Storage != StatusProblem { + t.Fatalf("storage status = %q, want %q", result.Storage, StatusProblem) + } + if !problemIssuesContain(result.Problems, "storage layout check failed") { + t.Fatalf("problems do not include storage layout failure: %v", result.Problems) + } + if !problemIssuesContain(result.Problems, filepath.Join("network-store", "coordinator")) { + t.Fatalf("problems do not mention missing coordinator network store: %v", result.Problems) + } +} + +func TestProbeBlocksWithStoreReportsCorruptBlockData(t *testing.T) { + expectedBlock := blocks.NewBlock([]byte("expected-data")) + returnedBlock, err := blocks.NewBlockWithCid([]byte("other-data"), expectedBlock.Cid()) + if err != nil { + t.Fatalf("make returned block: %v", err) + } + store := &doctorTestStore{ + block: returnedBlock, + } + inventory := Inventory{ + Files: []FileReport{{ + ID: "file1", + SpaceID: "space1", + Size: uint64(len(expectedBlock.RawData())), + CIDs: []string{expectedBlock.Cid().String()}, + CIDSizes: map[string]uint64{expectedBlock.Cid().String(): uint64(len(expectedBlock.RawData()))}, + }}, + } + + result, err := ProbeBlocksWithStore(context.Background(), inventory, store) + if err != nil { + t.Fatalf("ProbeBlocksWithStore() error = %v", err) + } + + key := fileReportKey("space1", "file1") + if len(result.CorruptByFile[key]) != 1 { + t.Fatalf("corrupt blocks for file = %v, want one block", result.CorruptByFile[key]) + } + if !problemIssuesContain(result.Problems, "block "+expectedBlock.Cid().String()+" content does not match its CID") { + t.Fatalf("problems do not include corrupt block content: %v", result.Problems) + } +} + +var _ filenodestore.Store = (*doctorTestStore)(nil) + +type doctorTestStore struct { + block blocks.Block +} + +func (s *doctorTestStore) Init(_ *app.App) error { + return nil +} + +func (s *doctorTestStore) Name() string { + return "doctor.test.store" +} + +func (s *doctorTestStore) Get(_ context.Context, _ cid.Cid) (blocks.Block, error) { + return s.block, nil +} + +func (s *doctorTestStore) GetMany(_ context.Context, _ []cid.Cid) <-chan blocks.Block { + ch := make(chan blocks.Block) + close(ch) + return ch +} + +func (s *doctorTestStore) Add(_ context.Context, _ []blocks.Block) error { + return nil +} + +func (s *doctorTestStore) Delete(_ context.Context, _ cid.Cid) error { + return nil +} + +func (s *doctorTestStore) DeleteMany(_ context.Context, _ []cid.Cid) error { + return nil +} + +func (s *doctorTestStore) IndexGet(_ context.Context, _ string) ([]byte, error) { + return nil, fileblockstore.ErrCIDNotFound +} + +func (s *doctorTestStore) IndexPut(_ context.Context, _ string, _ []byte) error { + return nil +} + +func (s *doctorTestStore) IndexDelete(_ context.Context, _ string) error { + return nil +} diff --git a/doctor/network.go b/doctor/network.go new file mode 100644 index 0000000..e7a5644 --- /dev/null +++ b/doctor/network.go @@ -0,0 +1,39 @@ +package doctor + +import ( + "context" + "fmt" + "net" + "time" +) + +func checkTCPListener(ctx context.Context, listenAddr string) error { + host, port, err := net.SplitHostPort(listenAddr) + if err != nil { + return fmt.Errorf("parse tcp listen address: %w", err) + } + dialHost := host + if host == "0.0.0.0" { + dialHost = "127.0.0.1" + } + if host == "::" { + dialHost = "::1" + } + + dialCtx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + + var dialer net.Dialer + conn, err := dialer.DialContext(dialCtx, "tcp", net.JoinHostPort(dialHost, port)) + if err != nil { + return err + } + return conn.Close() +} + +func checkUDPListener(listenAddr string) (HealthStatus, error) { + if _, _, err := net.SplitHostPort(listenAddr); err != nil { + return StatusProblem, fmt.Errorf("parse udp listen address: %w", err) + } + return StatusSkipped, nil +} diff --git a/doctor/network_test.go b/doctor/network_test.go new file mode 100644 index 0000000..e12d857 --- /dev/null +++ b/doctor/network_test.go @@ -0,0 +1,43 @@ +package doctor + +import ( + "context" + "net" + "testing" +) + +func TestCheckTCPListenerAcceptsRunningListener(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen tcp: %v", err) + } + defer listener.Close() + + err = checkTCPListener(context.Background(), listener.Addr().String()) + if err != nil { + t.Fatalf("checkTCPListener() error = %v", err) + } +} + +func TestCheckUDPListenerAcceptsBoundPort(t *testing.T) { + conn, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen udp: %v", err) + } + defer conn.Close() + + status, err := checkUDPListener(conn.LocalAddr().String()) + if err != nil { + t.Fatalf("checkUDPListener() error = %v", err) + } + if status != StatusSkipped { + t.Fatalf("udp status = %q, want %q", status, StatusSkipped) + } +} + +func TestCheckUDPListenerRejectsInvalidAddress(t *testing.T) { + _, err := checkUDPListener("not a hostport") + if err == nil { + t.Fatal("checkUDPListener() error = nil, want parse error") + } +} diff --git a/doctor/report.go b/doctor/report.go new file mode 100644 index 0000000..82fcb47 --- /dev/null +++ b/doctor/report.go @@ -0,0 +1,91 @@ +package doctor + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +const ( + reportDirectoryName = "doctor" + reportFileMode = 0o600 + reportDirMode = 0o750 +) + +func ReportPath(bundleConfigPath string, generatedAt time.Time) string { + name := fmt.Sprintf("doctor_%s.json", generatedAt.UTC().Format("2006-01-02T15-04-05Z")) + return filepath.Join(filepath.Dir(bundleConfigPath), reportDirectoryName, name) +} + +func SocketPath(bundleConfigPath string) string { + return filepath.Join(filepath.Dir(bundleConfigPath), "bundle.sock") +} + +func WriteReportAtomic(bundleConfigPath string, generatedAt time.Time, report Report) (string, error) { + path, err := nextReportPath(ReportPath(bundleConfigPath, generatedAt)) + if err != nil { + return "", err + } + dir := filepath.Dir(path) + if mkdirErr := os.MkdirAll(dir, reportDirMode); mkdirErr != nil { + return "", fmt.Errorf("create doctor report directory: %w", mkdirErr) + } + + tmpPath := filepath.Join(dir, "."+filepath.Base(path)+".tmp") + tmp, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, reportFileMode) + if err != nil { + return "", fmt.Errorf("create temporary report file: %w", err) + } + removeTmp := true + defer func() { + if removeTmp { + _ = os.Remove(tmpPath) + } + }() + + report.ReportPath = path + encoder := json.NewEncoder(tmp) + encoder.SetIndent("", " ") + if encodeErr := encoder.Encode(report); encodeErr != nil { + _ = tmp.Close() + return "", fmt.Errorf("encode doctor report: %w", encodeErr) + } + if chmodErr := tmp.Chmod(reportFileMode); chmodErr != nil { + _ = tmp.Close() + return "", fmt.Errorf("chmod doctor report: %w", chmodErr) + } + if closeErr := tmp.Close(); closeErr != nil { + return "", fmt.Errorf("close doctor report: %w", closeErr) + } + if renameErr := os.Rename(tmpPath, path); renameErr != nil { + return "", fmt.Errorf("publish doctor report: %w", renameErr) + } + + removeTmp = false + return path, nil +} + +func nextReportPath(path string) (string, error) { + if _, err := os.Stat(path); err != nil { + if errors.Is(err, os.ErrNotExist) { + return path, nil + } + return "", fmt.Errorf("check doctor report path: %w", err) + } + + extension := filepath.Ext(path) + stem := strings.TrimSuffix(path, extension) + for suffix := 2; ; suffix++ { + candidate := fmt.Sprintf("%s_%d%s", stem, suffix, extension) + if _, err := os.Stat(candidate); err != nil { + if errors.Is(err, os.ErrNotExist) { + return candidate, nil + } + return "", fmt.Errorf("check doctor report path: %w", err) + } + } +} diff --git a/doctor/report_test.go b/doctor/report_test.go new file mode 100644 index 0000000..58899a0 --- /dev/null +++ b/doctor/report_test.go @@ -0,0 +1,124 @@ +package doctor + +import ( + "encoding/json" + "os" + "path/filepath" + "runtime" + "testing" + "time" +) + +func TestReportPathUsesDoctorDirectoryNextToBundleConfig(t *testing.T) { + generatedAt := time.Date(2026, 5, 22, 14, 33, 10, 0, time.UTC) + bundleConfigPath := filepath.Join("/", "data", "bundle-config.yml") + + got := ReportPath(bundleConfigPath, generatedAt) + want := filepath.Join(filepath.Dir(bundleConfigPath), "doctor", "doctor_2026-05-22T14-33-10Z.json") + + if got != want { + t.Fatalf("ReportPath() = %q, want %q", got, want) + } +} + +func TestSocketPathUsesBundleConfigDirectory(t *testing.T) { + bundleConfigPath := filepath.Join("/", "data", "bundle-config.yml") + + got := SocketPath(bundleConfigPath) + want := filepath.Join(filepath.Dir(bundleConfigPath), "bundle.sock") + + if got != want { + t.Fatalf("SocketPath() = %q, want %q", got, want) + } +} + +func TestWriteReportAtomicCreatesDoctorDirectoryAndWritesJSON(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + generatedAt := time.Date(2026, 5, 22, 14, 33, 10, 0, time.UTC) + report := &Report{ + GeneratedAt: generatedAt, + BundleConfigPath: bundleConfigPath, + Verdict: VerdictHealthy, + Summary: Summary{ + Groups: 2, + Spaces: 13, + Files: 452, + CIDs: 12934, + }, + } + + path, err := WriteReportAtomic(bundleConfigPath, generatedAt, *report) + if err != nil { + t.Fatalf("WriteReportAtomic() error = %v", err) + } + + wantPath := filepath.Join(dir, "doctor", "doctor_2026-05-22T14-33-10Z.json") + if path != wantPath { + t.Fatalf("WriteReportAtomic() path = %q, want %q", path, wantPath) + } + + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read report: %v", err) + } + + var got Report + if unmarshalErr := json.Unmarshal(raw, &got); unmarshalErr != nil { + t.Fatalf("unmarshal report: %v", unmarshalErr) + } + if got.ReportPath != path { + t.Fatalf("report path in JSON = %q, want %q", got.ReportPath, path) + } + if got.Verdict != VerdictHealthy { + t.Fatalf("report verdict = %q, want %q", got.Verdict, VerdictHealthy) + } + if got.Summary.Spaces != 13 { + t.Fatalf("report spaces = %d, want 13", got.Summary.Spaces) + } + if runtime.GOOS != "windows" { + info, statErr := os.Stat(path) + if statErr != nil { + t.Fatalf("stat report: %v", statErr) + } + if info.Mode().Perm() != 0o600 { + t.Fatalf("report mode = %o, want 600", info.Mode().Perm()) + } + } + + tmpMatches, err := filepath.Glob(filepath.Join(dir, "doctor", "*.tmp")) + if err != nil { + t.Fatalf("glob tmp files: %v", err) + } + if len(tmpMatches) != 0 { + t.Fatalf("unexpected tmp files left behind: %v", tmpMatches) + } +} + +func TestWriteReportAtomicDoesNotOverwriteSameTimestampReport(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + generatedAt := time.Date(2026, 5, 22, 14, 33, 10, 0, time.UTC) + report := &Report{ + GeneratedAt: generatedAt, + Verdict: VerdictHealthy, + } + + firstPath, err := WriteReportAtomic(bundleConfigPath, generatedAt, *report) + if err != nil { + t.Fatalf("first WriteReportAtomic() error = %v", err) + } + secondPath, err := WriteReportAtomic(bundleConfigPath, generatedAt, *report) + if err != nil { + t.Fatalf("second WriteReportAtomic() error = %v", err) + } + + if firstPath == secondPath { + t.Fatalf("second report overwrote first path %q", firstPath) + } + for _, path := range []string{firstPath, secondPath} { + if _, statErr := os.Stat(path); statErr != nil { + t.Fatalf("report %s does not exist: %v", path, statErr) + } + } +} diff --git a/doctor/runner.go b/doctor/runner.go new file mode 100644 index 0000000..86d80dc --- /dev/null +++ b/doctor/runner.go @@ -0,0 +1,548 @@ +package doctor + +import ( + "context" + "errors" + "fmt" + "io" + "net/url" + "os" + "slices" + "time" + + "github.com/anyproto/any-sync/nodeconf" + "gopkg.in/yaml.v3" + + bundleconfig "github.com/grishy/any-sync-bundle/config" +) + +type RuntimeRunnerConfig struct { + BundleConfig *bundleconfig.Config + BundleConfigPath string + ClientConfigPath string + Build BuildInfo + Now func() time.Time + + LoadIndexSnapshot func(ctx context.Context) (IndexSnapshot, error) + ProbeBlocks func(ctx context.Context, inventory Inventory) (BlockProbeResult, error) + ProbeRuntime func(ctx context.Context) RuntimeProbeResult +} + +type RuntimeRunner struct { + cfg RuntimeRunnerConfig +} + +type RuntimeProbeResult struct { + Mongo HealthStatus + Redis HealthStatus + RedisBloom HealthStatus + Storage HealthStatus + Problems []Problem +} + +type BlockProbeResult struct { + Checked uint64 + MissingByFile map[string][]string + CorruptByFile map[string][]string + Problems []Problem +} + +func NewRuntimeRunner(cfg RuntimeRunnerConfig) *RuntimeRunner { + return &RuntimeRunner{cfg: cfg} +} + +//nolint:funlen // The output is a fixed seven-phase diagnostic transcript; splitting would hide ordering. +func (r *RuntimeRunner) RunDoctor(ctx context.Context, out io.Writer) (*Report, error) { + if err := r.validate(); err != nil { + return nil, err + } + + generatedAt := r.cfg.Now().UTC() + report := &Report{ + GeneratedAt: generatedAt, + Experimental: true, + ReportSchema: currentReportSchema, + Build: r.cfg.Build, + BundleConfigPath: r.cfg.BundleConfigPath, + ClientConfigPath: r.cfg.ClientConfigPath, + Config: buildConfigReport(r.cfg.BundleConfig), + Verdict: VerdictHealthy, + } + problems := []Problem{} + + fmt.Fprintln(out, "Experimental:") + fmt.Fprintln(out, " doctor output and JSON report schema may change between releases.") + fmt.Fprintln(out) + + fmt.Fprintln(out, "[1/7] Config") + configProblems := r.checkConfig() + if len(configProblems) == 0 { + fmt.Fprintln(out, " status: ok") + } else { + fmt.Fprintln(out, " status: problem") + } + for _, problem := range configProblems { + fmt.Fprintf(out, " - %s\n", problem.Issue) + } + problems = append(problems, configProblems...) + fmt.Fprintln(out) + + fmt.Fprintln(out, "[2/7] Runtime") + runtimeResult := r.cfg.ProbeRuntime(ctx) + report.Runtime = RuntimeReport{ + Mongo: runtimeResult.Mongo, + Redis: runtimeResult.Redis, + RedisBloom: runtimeResult.RedisBloom, + Storage: runtimeResult.Storage, + } + fmt.Fprintf(out, " mongo: %s\n", runtimeResult.Mongo) + fmt.Fprintf(out, " redis: %s\n", runtimeResult.Redis) + fmt.Fprintf(out, " redis bloom: %s\n", runtimeResult.RedisBloom) + fmt.Fprintf(out, " storage: %s\n\n", runtimeResult.Storage) + problems = append(problems, runtimeResult.Problems...) + if err := runtimePrerequisiteError(runtimeResult); err != nil { + return nil, err + } + + fmt.Fprintln(out, "[3/7] Network") + networkReport, networkProblems := r.checkNetwork(ctx) + report.Network = networkReport + fmt.Fprintf(out, " tcp listen: %s\n", networkReport.TCPListen) + fmt.Fprintf(out, " udp listen: %s\n", networkReport.UDPListen) + fmt.Fprintf(out, " advertised addresses: %s\n", networkReport.Advertised) + for _, problem := range networkProblems { + fmt.Fprintf(out, " - %s\n", problem.Issue) + } + problems = append(problems, networkProblems...) + fmt.Fprintln(out) + + fmt.Fprintln(out, "[4/7] Inventory") + snapshot, err := r.cfg.LoadIndexSnapshot(ctx) + if err != nil { + return nil, fmt.Errorf("load filenode index snapshot: %w", err) + } + inventory, indexProblems, err := InspectIndexSnapshot(snapshot) + if err != nil { + return nil, fmt.Errorf("inspect filenode index snapshot: %w", err) + } + problems = append(problems, indexProblems...) + printInventory(out, inventory) + fmt.Fprintln(out) + + fmt.Fprintln(out, "[5/7] Spaces and index") + printSpaces(out, inventory) + fmt.Fprintf(out, " checked files: %d\n", len(inventory.Files)) + fmt.Fprintf(out, " checked cid refs: %d\n", countFileCIDRefs(inventory.Files)) + fmt.Fprintf(out, " problems: %d\n\n", len(indexProblems)) + + fmt.Fprintln(out, "[6/7] Blocks") + blockResult, err := r.cfg.ProbeBlocks(ctx, inventory) + if err != nil { + return nil, fmt.Errorf("probe filenode blocks: %w", err) + } + blockProblems := applyBlockProbeResult(&inventory, blockResult) + blockProblems = append(blockProblems, blockResult.Problems...) + problems = append(problems, blockProblems...) + fmt.Fprintf(out, " checked blocks: %d\n", blockResult.Checked) + fmt.Fprintf(out, " missing blocks: %d\n", countMissingBlocks(blockResult)) + fmt.Fprintf(out, " corrupt blocks: %d\n\n", countCorruptBlocks(blockResult)) + + report.Inventory = inventory + report.Summary = summarizeInventory(inventory) + report.Problems = problems + report.Verdict, report.SuggestedNextAction = classifyVerdict(inventory, problems) + + fmt.Fprintln(out, "[7/7] Report") + path, err := WriteReportAtomic(r.cfg.BundleConfigPath, generatedAt, *report) + if err != nil { + return nil, err + } + report.ReportPath = path + fmt.Fprintf(out, " written: %s\n\n", path) + printProblems(out, problems) + fmt.Fprintln(out, "Verdict:") + fmt.Fprintf(out, " %s\n\n", report.Verdict) + if report.SuggestedNextAction != "" { + fmt.Fprintln(out, "Suggested next action:") + fmt.Fprintf(out, " %s\n\n", report.SuggestedNextAction) + } + fmt.Fprintln(out, "Report written:") + fmt.Fprintf(out, " %s\n", path) + + return report, nil +} + +func (r *RuntimeRunner) validate() error { + if r.cfg.BundleConfig == nil { + return errors.New("doctor bundle config is required") + } + if r.cfg.BundleConfigPath == "" { + return errors.New("doctor bundle config path is required") + } + if r.cfg.ClientConfigPath == "" { + return errors.New("doctor client config path is required") + } + if r.cfg.Now == nil { + return errors.New("doctor clock is required") + } + if r.cfg.ProbeRuntime == nil { + return errors.New("doctor runtime probe is required") + } + if r.cfg.LoadIndexSnapshot == nil { + return errors.New("doctor index snapshot loader is required") + } + if r.cfg.ProbeBlocks == nil { + return errors.New("doctor block probe is required") + } + return nil +} + +func runtimePrerequisiteError(result RuntimeProbeResult) error { + if result.Redis == StatusProblem { + return errors.New("redis is required for filenode inventory scan; fix Redis and run doctor again") + } + return nil +} + +func classifyVerdict(inventory Inventory, problems []Problem) (Verdict, string) { + if len(problems) == 0 { + return VerdictHealthy, "No action needed." + } + for _, file := range inventory.Files { + if len(file.MissingBlocks) > 0 { + return VerdictFilesRequireClientReupload, + "Missing blocks cannot be recreated by the server; re-upload from the original client/cache." + } + if len(file.CorruptBlocks) > 0 { + return VerdictFilesRequireClientReupload, + "Corrupt blocks cannot be trusted by the server; re-upload from the original client/cache." + } + } + return VerdictProblemsFound, "Review the problem list and JSON report before changing data." +} + +func (r *RuntimeRunner) checkConfig() []Problem { + problems := []Problem{} + if err := r.cfg.BundleConfig.Validate(); err != nil { + problems = append(problems, Problem{ + Scope: problemScopeConfig, + Issue: fmt.Sprintf("bundle config is invalid: %v", err), + }) + } + // #nosec G703 -- Doctor reads the local client config path selected by the running bundle. + if _, statErr := os.Stat(r.cfg.ClientConfigPath); statErr != nil { + problems = append(problems, Problem{ + Scope: problemScopeConfig, + Issue: fmt.Sprintf("client config cannot be read: %v", statErr), + }) + } else { + matchErr := checkClientConfigMatches(r.cfg.BundleConfig, r.cfg.ClientConfigPath) + if matchErr != nil { + problems = append(problems, Problem{ + Scope: problemScopeConfig, + Issue: fmt.Sprintf("client config does not match current bundle config: %v", matchErr), + }) + } + } + return problems +} + +func checkClientConfigMatches(cfg *bundleconfig.Config, path string) error { + expectedData, err := cfg.YamlClientConfig() + if err != nil { + return fmt.Errorf("generate expected client config: %w", err) + } + // #nosec G703 -- Doctor compares the local client config path selected by the running bundle. + actualData, err := os.ReadFile(path) + if err != nil { + return err + } + + var expected nodeconf.Configuration + if decodeExpectedErr := yaml.Unmarshal(expectedData, &expected); decodeExpectedErr != nil { + return fmt.Errorf("decode expected client config: %w", decodeExpectedErr) + } + var actual nodeconf.Configuration + if decodeActualErr := yaml.Unmarshal(actualData, &actual); decodeActualErr != nil { + return fmt.Errorf("decode current client config: %w", decodeActualErr) + } + + if actual.Id != expected.Id { + return fmt.Errorf("config id: current=%q expected=%q", actual.Id, expected.Id) + } + if actual.NetworkId != expected.NetworkId { + return fmt.Errorf("network id: current=%q expected=%q", actual.NetworkId, expected.NetworkId) + } + if len(actual.Nodes) != len(expected.Nodes) { + return fmt.Errorf("node count: current=%d expected=%d", len(actual.Nodes), len(expected.Nodes)) + } + for idx := range expected.Nodes { + actualNode := actual.Nodes[idx] + expectedNode := expected.Nodes[idx] + if actualNode.PeerId != expectedNode.PeerId { + return fmt.Errorf("node[%d] peer id: current=%q expected=%q", idx, actualNode.PeerId, expectedNode.PeerId) + } + if !slices.Equal(actualNode.Addresses, expectedNode.Addresses) { + return fmt.Errorf( + "node[%d] addresses: current=%v expected=%v", + idx, + actualNode.Addresses, + expectedNode.Addresses, + ) + } + if !slices.Equal(actualNode.Types, expectedNode.Types) { + return fmt.Errorf("node[%d] types: current=%v expected=%v", idx, actualNode.Types, expectedNode.Types) + } + } + return nil +} + +func (r *RuntimeRunner) checkNetwork(ctx context.Context) (NetworkReport, []Problem) { + report := buildNetworkReport(r.cfg.BundleConfig) + problems := []Problem{} + if err := checkTCPListener(ctx, r.cfg.BundleConfig.Network.ListenTCPAddr); err != nil { + report.TCPListen = StatusProblem + problems = append(problems, Problem{ + Scope: problemScopeNetwork, + Issue: fmt.Sprintf("tcp listener check failed: %v", err), + }) + } else { + report.TCPListen = StatusOK + } + udpStatus, err := checkUDPListener(r.cfg.BundleConfig.Network.ListenUDPAddr) + if err != nil { + report.UDPListen = StatusProblem + problems = append(problems, Problem{ + Scope: problemScopeNetwork, + Issue: fmt.Sprintf("udp listener check failed: %v", err), + }) + } else { + report.UDPListen = udpStatus + } + if len(r.cfg.BundleConfig.ExternalAddr) == 0 { + report.Advertised = StatusProblem + problems = append(problems, Problem{ + Scope: problemScopeNetwork, + Issue: "no advertised addresses configured", + }) + } else { + report.Advertised = StatusOK + } + return report, problems +} + +func buildConfigReport(cfg *bundleconfig.Config) ConfigReport { + report := ConfigReport{ + ConfigID: cfg.ConfigID, + NetworkID: cfg.NetworkID, + PeerID: cfg.Account.PeerId, + MongoCoordinatorURI: redactedURI(cfg.Coordinator.MongoConnect), + MongoConsensusURI: redactedURI(cfg.Consensus.MongoConnect), + RedisURI: redactedURI(cfg.FileNode.RedisConnect), + StoragePath: cfg.StoragePath, + AdvertisedAddresses: append([]string(nil), cfg.ExternalAddr...), + } + if cfg.FileNode.S3 != nil { + report.S3 = &S3Report{ + Bucket: cfg.FileNode.S3.Bucket, + Endpoint: redactedURI(cfg.FileNode.S3.Endpoint), + Region: cfg.FileNode.S3.Region, + ForcePathStyle: cfg.FileNode.S3.ForcePathStyle, + } + } + return report +} + +func buildNetworkReport(cfg *bundleconfig.Config) NetworkReport { + return NetworkReport{ + ListenTCPAddr: cfg.Network.ListenTCPAddr, + ListenUDPAddr: cfg.Network.ListenUDPAddr, + AdvertisedAddresses: append([]string(nil), cfg.ExternalAddr...), + } +} + +func redactedURI(raw string) string { + parsed, err := url.Parse(raw) + if err != nil { + return "" + } + parsed.User = nil + parsed.RawQuery = "" + parsed.Fragment = "" + return parsed.String() +} + +func printInventory(out io.Writer, inventory Inventory) { + fmt.Fprintf(out, " accounts/groups: %d\n", len(inventory.Groups)) + for _, group := range inventory.Groups { + fmt.Fprintf(out, " - %s spaces=%d files=%d cids=%d bytes=%d status=%s\n", + group.ID, group.Spaces, group.Files, group.CIDs, group.Bytes, group.Status) + } + fmt.Fprintln(out) + fmt.Fprintf(out, " spaces: %d\n", len(inventory.Spaces)) + for _, space := range inventory.Spaces { + fmt.Fprintf(out, " - %s account=%s files=%d cids=%d bytes=%d status=%s\n", + space.ID, space.GroupID, space.Files, space.CIDs, space.Bytes, space.Status) + } + fmt.Fprintln(out) + fmt.Fprintf(out, " files: %d\n", len(inventory.Files)) + fmt.Fprintf(out, " cids: %d\n", countUniqueFileCIDs(inventory.Files)) +} + +func printSpaces(out io.Writer, inventory Inventory) { + for _, space := range inventory.Spaces { + fmt.Fprintf(out, " space %s\n", space.ID) + fmt.Fprintf(out, " files: %d\n", space.Files) + fmt.Fprintf(out, " cids: %d\n", space.CIDs) + if space.MissingCIDIndex > 0 { + fmt.Fprintf(out, " index: missing %d cid entries\n", space.MissingCIDIndex) + } else { + if space.IndexProblems > 0 { + fmt.Fprintf(out, " index: problems %d\n", space.IndexProblems) + } else { + fmt.Fprintln(out, " index: ok") + } + } + if space.MissingBlocks == 0 { + if space.CIDs == 0 { + fmt.Fprintln(out, " blocks: none") + } else { + fmt.Fprintln(out, " blocks: ok") + } + } else { + fmt.Fprintf(out, " blocks: missing %d\n", space.MissingBlocks) + } + fmt.Fprintln(out) + } +} + +func printProblems(out io.Writer, problems []Problem) { + if len(problems) == 0 { + fmt.Fprintln(out, "Problems:") + fmt.Fprintln(out, " none") + fmt.Fprintln(out) + return + } + fmt.Fprintln(out, "Problems:") + for idx, problem := range problems { + fmt.Fprintf(out, " %d. %s", idx+1, problem.Scope) + if problem.ID != "" { + fmt.Fprintf(out, " %s", problem.ID) + } + fmt.Fprintln(out) + fmt.Fprintf(out, " issue: %s\n", problem.Issue) + if problem.Recoverable != "" { + fmt.Fprintf(out, " recoverable: %s\n", problem.Recoverable) + } + } + fmt.Fprintln(out) +} + +func applyBlockProbeResult(inventory *Inventory, result BlockProbeResult) []Problem { + if len(result.MissingByFile) == 0 && len(result.CorruptByFile) == 0 { + return nil + } + filesByKey := map[string]*FileReport{} + spacesByID := map[string]*SpaceReport{} + for idx := range inventory.Files { + file := &inventory.Files[idx] + filesByKey[fileReportKey(file.SpaceID, file.ID)] = file + } + for idx := range inventory.Spaces { + space := &inventory.Spaces[idx] + spacesByID[space.ID] = space + } + + problems := []Problem{} + for key, cids := range result.MissingByFile { + file := filesByKey[key] + if file == nil { + continue + } + file.MissingBlocks = append(file.MissingBlocks, cids...) + space := spacesByID[file.SpaceID] + if space != nil { + space.MissingBlocks += uint64(len(cids)) + space.Status = StatusProblem + } + problems = append(problems, Problem{ + Scope: "file", + ID: file.ID, + Issue: fmt.Sprintf("%d referenced blocks are missing", len(cids)), + Recoverable: "requires original client/cache", + }) + } + for key, cids := range result.CorruptByFile { + file := filesByKey[key] + if file == nil { + continue + } + file.CorruptBlocks = append(file.CorruptBlocks, cids...) + space := spacesByID[file.SpaceID] + if space != nil { + space.CorruptBlocks += uint64(len(cids)) + space.Status = StatusProblem + } + problems = append(problems, Problem{ + Scope: "file", + ID: file.ID, + Issue: fmt.Sprintf("%d referenced blocks are corrupt", len(cids)), + Recoverable: "requires original client/cache", + }) + } + return problems +} + +func summarizeInventory(inventory Inventory) Summary { + var bytes uint64 + var cids uint64 + for _, space := range inventory.Spaces { + bytes += space.Bytes + cids += space.CIDs + } + return Summary{ + Groups: uint64(len(inventory.Groups)), + Spaces: uint64(len(inventory.Spaces)), + Files: uint64(len(inventory.Files)), + CIDs: cids, + Bytes: bytes, + } +} + +func countFileCIDRefs(files []FileReport) uint64 { + var count uint64 + for _, file := range files { + count += uint64(len(file.CIDs)) + } + return count +} + +func countUniqueFileCIDs(files []FileReport) uint64 { + seen := map[string]struct{}{} + for _, file := range files { + for _, cid := range file.CIDs { + seen[cid] = struct{}{} + } + } + return uint64(len(seen)) +} + +func countMissingBlocks(result BlockProbeResult) uint64 { + var count uint64 + for _, cids := range result.MissingByFile { + count += uint64(len(cids)) + } + return count +} + +func countCorruptBlocks(result BlockProbeResult) uint64 { + var count uint64 + for _, cids := range result.CorruptByFile { + count += uint64(len(cids)) + } + return count +} + +func fileReportKey(spaceID string, fileID string) string { + return spaceID + "\x00" + fileID +} diff --git a/doctor/runner_test.go b/doctor/runner_test.go new file mode 100644 index 0000000..daa4cb6 --- /dev/null +++ b/doctor/runner_test.go @@ -0,0 +1,513 @@ +package doctor + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/anyproto/any-sync-filenode/index" + "github.com/anyproto/any-sync-filenode/index/indexproto" + "github.com/anyproto/any-sync/accountservice" + + bundleconfig "github.com/grishy/any-sync-bundle/config" +) + +func TestRuntimeRunnerWritesReportAndReturnsProblemVerdict(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + generatedAt := time.Date(2026, 5, 22, 14, 33, 10, 0, time.UTC) + cfg := validRuntimeTestConfig(t) + writeRuntimeTestClientConfig(t, cfg, clientConfigPath) + groupID := "account1" + spaceID := "space1" + fileID := "file1" + cidMissing := "bafy-missing" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + runner := NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg, + BundleConfigPath: bundleConfigPath, + ClientConfigPath: clientConfigPath, + Build: BuildInfo{ + Version: "test-version", + Commit: "test-commit", + Date: "2026-05-22", + }, + Now: func() time.Time { return generatedAt }, + LoadIndexSnapshot: func(_ context.Context) (IndexSnapshot, error) { + return IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}}, + ), + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey(fileID): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidMissing}, Size: 10}, + ), + }, + }, + }, nil + }, + ProbeBlocks: func(_ context.Context, _ Inventory) (BlockProbeResult, error) { + return BlockProbeResult{Checked: 1}, nil + }, + ProbeRuntime: healthyRuntimeProbe, + }) + + var out bytes.Buffer + report, err := runner.RunDoctor(context.Background(), &out) + if err != nil { + t.Fatalf("RunDoctor() error = %v", err) + } + + if report.Verdict != VerdictProblemsFound { + t.Fatalf("verdict = %q, want %q", report.Verdict, VerdictProblemsFound) + } + if report.ReportPath != filepath.Join(dir, "doctor", "doctor_2026-05-22T14-33-10Z.json") { + t.Fatalf("report path = %q", report.ReportPath) + } + + output := out.String() + for _, want := range []string{ + "Experimental:", + "doctor output and JSON report schema may change between releases", + "[7/7] Report", + "[1/7] Config", + "[4/7] Inventory", + "space space1", + "Problems:", + "Verdict:", + } { + if !strings.Contains(output, want) { + t.Fatalf("output does not contain %q:\n%s", want, output) + } + } + + raw, err := os.ReadFile(report.ReportPath) + if err != nil { + t.Fatalf("read report: %v", err) + } + var written Report + if unmarshalErr := json.Unmarshal(raw, &written); unmarshalErr != nil { + t.Fatalf("unmarshal report: %v", unmarshalErr) + } + if written.Summary.Spaces != 1 { + t.Fatalf("written spaces = %d, want 1", written.Summary.Spaces) + } + if written.Build.Version != "test-version" { + t.Fatalf("written build version = %q, want test-version", written.Build.Version) + } + if !written.Experimental { + t.Fatal("written experimental = false, want true") + } + if written.ReportSchema != currentReportSchema { + t.Fatalf("written report schema = %d, want %d", written.ReportSchema, currentReportSchema) + } + if len(written.Problems) == 0 { + t.Fatal("written problems = 0, want at least one problem") + } + if !problemIssuesContain(written.Problems, "global index entry is missing") { + t.Fatalf("written problems do not include missing global index entry: %v", written.Problems) + } +} + +func TestRuntimeRunnerWritesRedactedConfigAndRuntimeDetails(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + cfg := validRuntimeTestConfig(t) + cfg.ConfigID = "config-id" + cfg.NetworkID = "network-id" + cfg.Account.PeerId = "peer-id" + cfg.Account.PeerKey = "peer-secret" + cfg.Account.SigningKey = "sign-secret" + cfg.ExternalAddr = []string{"sync.example.com"} + cfg.Coordinator.MongoConnect = "mongodb://mongo-user:mongo-secret@127.0.0.1:27017/?authSource=admin" + cfg.Consensus.MongoConnect = "mongodb://consensus-user:consensus-secret@127.0.0.1:27017/?w=majority" + cfg.FileNode.RedisConnect = "redis://:redis-secret@127.0.0.1:6379/1" + cfg.FileNode.S3 = &bundleconfig.S3Config{ + Bucket: "anytype-data", + Endpoint: "https://s3-user:s3-secret@s3.example.com/bucket?token=s3-token#frag", + Region: "eu-central-1", + ForcePathStyle: true, + } + writeRuntimeTestClientConfig(t, cfg, clientConfigPath) + runner := NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg, + BundleConfigPath: bundleConfigPath, + ClientConfigPath: clientConfigPath, + Now: fixedRuntimeTestNow, + LoadIndexSnapshot: func(_ context.Context) (IndexSnapshot, error) { + return IndexSnapshot{}, nil + }, + ProbeBlocks: func(_ context.Context, _ Inventory) (BlockProbeResult, error) { + return BlockProbeResult{}, nil + }, + ProbeRuntime: healthyRuntimeProbe, + }) + + var out bytes.Buffer + report, err := runner.RunDoctor(context.Background(), &out) + if err != nil { + t.Fatalf("RunDoctor() error = %v", err) + } + + raw, err := os.ReadFile(report.ReportPath) + if err != nil { + t.Fatalf("read report: %v", err) + } + for _, secret := range []string{ + "mongo-secret", + "consensus-secret", + "redis-secret", + "peer-secret", + "sign-secret", + "s3-secret", + "s3-token", + } { + if bytes.Contains(raw, []byte(secret)) { + t.Fatalf("report contains secret %q:\n%s", secret, string(raw)) + } + } + + var written Report + if unmarshalErr := json.Unmarshal(raw, &written); unmarshalErr != nil { + t.Fatalf("unmarshal report: %v", unmarshalErr) + } + if written.Config.ConfigID != "config-id" { + t.Fatalf("config ID = %q, want config-id", written.Config.ConfigID) + } + if written.Config.NetworkID != "network-id" { + t.Fatalf("network ID = %q, want network-id", written.Config.NetworkID) + } + if written.Config.PeerID != "peer-id" { + t.Fatalf("peer ID = %q, want peer-id", written.Config.PeerID) + } + if written.Config.MongoCoordinatorURI != "mongodb://127.0.0.1:27017/" { + t.Fatalf("coordinator URI = %q", written.Config.MongoCoordinatorURI) + } + if written.Config.RedisURI != "redis://127.0.0.1:6379/1" { + t.Fatalf("redis URI = %q", written.Config.RedisURI) + } + if written.Config.S3 == nil { + t.Fatal("s3 config is nil") + } + if written.Config.S3.Bucket != "anytype-data" { + t.Fatalf("s3 bucket = %q, want anytype-data", written.Config.S3.Bucket) + } + if written.Config.S3.Endpoint != "https://s3.example.com/bucket" { + t.Fatalf("s3 endpoint = %q", written.Config.S3.Endpoint) + } + if written.Network.ListenTCPAddr != cfg.Network.ListenTCPAddr { + t.Fatalf("tcp listen = %q, want %q", written.Network.ListenTCPAddr, cfg.Network.ListenTCPAddr) + } + if written.Runtime.Storage != StatusOK { + t.Fatalf("storage status = %q, want ok", written.Runtime.Storage) + } +} + +func TestRuntimeRunnerReportsStaleClientConfig(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + cfg := validRuntimeTestConfig(t) + cfg.NetworkID = "network-id" + staleClientConfig := `id: config-id +networkId: stale-network-id +nodes: + - peerId: peer-id + addresses: [] + types: [] +` + if err := os.WriteFile(clientConfigPath, []byte(staleClientConfig), 0o644); err != nil { + t.Fatalf("write stale client config: %v", err) + } + runner := NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg, + BundleConfigPath: bundleConfigPath, + ClientConfigPath: clientConfigPath, + Now: fixedRuntimeTestNow, + LoadIndexSnapshot: func(_ context.Context) (IndexSnapshot, error) { + return IndexSnapshot{}, nil + }, + ProbeBlocks: func(_ context.Context, _ Inventory) (BlockProbeResult, error) { + return BlockProbeResult{}, nil + }, + ProbeRuntime: healthyRuntimeProbe, + }) + + var out bytes.Buffer + report, err := runner.RunDoctor(context.Background(), &out) + if err != nil { + t.Fatalf("RunDoctor() error = %v", err) + } + + if !problemIssuesContain(report.Problems, "client config does not match current bundle config") { + t.Fatalf("problems do not include stale client config: %v", report.Problems) + } +} + +func TestRuntimeRunnerStopsBeforeInventoryWhenRedisIsUnavailable(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + cfg := validRuntimeTestConfig(t) + writeRuntimeTestClientConfig(t, cfg, clientConfigPath) + runner := NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg, + BundleConfigPath: bundleConfigPath, + ClientConfigPath: clientConfigPath, + Now: fixedRuntimeTestNow, + LoadIndexSnapshot: func(_ context.Context) (IndexSnapshot, error) { + t.Fatal("LoadIndexSnapshot was called even though Redis is unavailable") + return IndexSnapshot{}, nil + }, + ProbeBlocks: func(_ context.Context, _ Inventory) (BlockProbeResult, error) { + t.Fatal("ProbeBlocks was called even though Redis is unavailable") + return BlockProbeResult{}, nil + }, + ProbeRuntime: func(_ context.Context) RuntimeProbeResult { + return RuntimeProbeResult{ + Mongo: StatusOK, + Redis: StatusProblem, + RedisBloom: StatusProblem, + Storage: StatusOK, + Problems: []Problem{{ + Scope: problemScopeRuntime, + Issue: "redis ping failed: connection refused", + }}, + } + }, + }) + + var out bytes.Buffer + report, err := runner.RunDoctor(context.Background(), &out) + if err == nil { + t.Fatal("RunDoctor() error = nil, want Redis prerequisite error") + } + if report != nil { + t.Fatalf("report = %#v, want nil", report) + } + if !strings.Contains(err.Error(), "redis is required for filenode inventory scan") { + t.Fatalf("RunDoctor() error = %v", err) + } + if strings.Contains(out.String(), "[3/7] Network") { + t.Fatalf("doctor continued after Redis prerequisite failure:\n%s", out.String()) + } + if strings.Contains(out.String(), "Report:") { + t.Fatalf("doctor printed report path even though no report was written:\n%s", out.String()) + } + if _, statErr := os.Stat(filepath.Join(dir, "doctor")); !os.IsNotExist(statErr) { + t.Fatalf("doctor report directory exists after prerequisite failure: %v", statErr) + } +} + +func TestRuntimeRunnerDoesNotPrintReportPathWhenInventoryFails(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + cfg := validRuntimeTestConfig(t) + writeRuntimeTestClientConfig(t, cfg, clientConfigPath) + runner := NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg, + BundleConfigPath: bundleConfigPath, + ClientConfigPath: clientConfigPath, + Now: fixedRuntimeTestNow, + LoadIndexSnapshot: func(_ context.Context) (IndexSnapshot, error) { + return IndexSnapshot{}, errors.New("redis scan interrupted") + }, + ProbeBlocks: func(_ context.Context, _ Inventory) (BlockProbeResult, error) { + t.Fatal("ProbeBlocks was called even though inventory failed") + return BlockProbeResult{}, nil + }, + ProbeRuntime: healthyRuntimeProbe, + }) + + var out bytes.Buffer + _, err := runner.RunDoctor(context.Background(), &out) + if err == nil { + t.Fatal("RunDoctor() error = nil, want inventory error") + } + if strings.Contains(out.String(), "/doctor/doctor_") { + t.Fatalf("doctor printed report path even though no report was written:\n%s", out.String()) + } +} + +func TestRuntimeRunnerAddsMissingBlockProblems(t *testing.T) { + dir := t.TempDir() + bundleConfigPath := filepath.Join(dir, "bundle-config.yml") + clientConfigPath := filepath.Join(dir, "client-config.yml") + cfg := validRuntimeTestConfig(t) + writeRuntimeTestClientConfig(t, cfg, clientConfigPath) + groupID := "account1" + spaceID := "space1" + fileID := "file1" + cidMissingBlock := "bafy-missing-block" + key := index.Key{GroupId: groupID, SpaceId: spaceID} + runner := NewRuntimeRunner(RuntimeRunnerConfig{ + BundleConfig: cfg, + BundleConfigPath: bundleConfigPath, + ClientConfigPath: clientConfigPath, + Now: fixedRuntimeTestNow, + LoadIndexSnapshot: func(_ context.Context) (IndexSnapshot, error) { + return IndexSnapshot{ + Hashes: map[string]map[string]string{ + index.GroupKey(key): { + redisIndexInfoField: mustMarshalGroupEntry( + t, + &indexproto.GroupEntry{GroupId: groupID, SpaceIds: []string{spaceID}}, + ), + }, + index.SpaceKey(key): { + redisIndexInfoField: mustMarshalSpaceEntry( + t, + &indexproto.SpaceEntry{GroupId: groupID, FileCount: 1, CidCount: 1, Size: 10}, + ), + index.FileKey(fileID): mustMarshalFileEntry( + t, + &indexproto.FileEntry{Cids: []string{cidMissingBlock}, Size: 10}, + ), + }, + }, + Values: map[string]string{ + redisIndexCIDKey(cidMissingBlock): mustMarshalCIDEntry(t, &indexproto.CidEntry{Size: 10, Refs: 1}), + }, + }, nil + }, + ProbeBlocks: func(_ context.Context, _ Inventory) (BlockProbeResult, error) { + return BlockProbeResult{ + Checked: 1, + MissingByFile: map[string][]string{ + fileReportKey(spaceID, fileID): {cidMissingBlock}, + }, + }, nil + }, + ProbeRuntime: healthyRuntimeProbe, + }) + + var out bytes.Buffer + report, err := runner.RunDoctor(context.Background(), &out) + if err != nil { + t.Fatalf("RunDoctor() error = %v", err) + } + + if report.Summary.Spaces != 1 { + t.Fatalf("spaces = %d, want 1", report.Summary.Spaces) + } + if report.Inventory.Spaces[0].MissingBlocks != 1 { + t.Fatalf("missing blocks = %d, want 1", report.Inventory.Spaces[0].MissingBlocks) + } + if report.Verdict != VerdictFilesRequireClientReupload { + t.Fatalf("verdict = %q, want %q", report.Verdict, VerdictFilesRequireClientReupload) + } + if !strings.Contains(report.SuggestedNextAction, "original client/cache") { + t.Fatalf("suggested next action = %q", report.SuggestedNextAction) + } + if !strings.Contains(out.String(), "missing blocks: 1") { + t.Fatalf("output does not mention missing block:\n%s", out.String()) + } +} + +func TestPrintSpacesShowsIndexProblemsWithoutMissingGlobalCID(t *testing.T) { + inventory := Inventory{ + Spaces: []SpaceReport{{ + ID: "space1", + Files: 1, + CIDs: 1, + IndexProblems: 2, + Status: StatusProblem, + }}, + } + + var out bytes.Buffer + printSpaces(&out, inventory) + + if !strings.Contains(out.String(), "index: problems 2") { + t.Fatalf("output does not show index problems:\n%s", out.String()) + } +} + +func validRuntimeTestConfig(t *testing.T) *bundleconfig.Config { + t.Helper() + + tcpListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen tcp: %v", err) + } + t.Cleanup(func() { + _ = tcpListener.Close() + }) + udpListener, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen udp: %v", err) + } + t.Cleanup(func() { + _ = udpListener.Close() + }) + + return &bundleconfig.Config{ + ExternalAddr: []string{"127.0.0.1"}, + ConfigID: "config-id", + NetworkID: "network-id", + StoragePath: "./data/storage", + Account: accountservice.Config{ + PeerId: "peer-id", + }, + Network: bundleconfig.NetworkConfig{ + ListenTCPAddr: tcpListener.Addr().String(), + ListenUDPAddr: udpListener.LocalAddr().String(), + }, + Coordinator: bundleconfig.CoordinatorConfig{ + MongoConnect: "mongodb://127.0.0.1:27017/", + MongoDatabase: "coordinator", + }, + Consensus: bundleconfig.ConsensusConfig{ + MongoConnect: "mongodb://127.0.0.1:27017/?w=majority", + MongoDatabase: "consensus", + }, + FileNode: bundleconfig.FileNodeConfig{ + RedisConnect: "redis://127.0.0.1:6379/", + DefaultLimit: 1, + }, + } +} + +func writeRuntimeTestClientConfig(t *testing.T, cfg *bundleconfig.Config, path string) { + t.Helper() + + data, err := cfg.YamlClientConfig() + if err != nil { + t.Fatalf("client config yaml: %v", err) + } + if writeErr := os.WriteFile(path, data, 0o644); writeErr != nil { + t.Fatalf("write client config: %v", writeErr) + } +} + +func healthyRuntimeProbe(_ context.Context) RuntimeProbeResult { + return RuntimeProbeResult{ + Mongo: StatusOK, + Redis: StatusOK, + RedisBloom: StatusOK, + Storage: StatusOK, + } +} + +func fixedRuntimeTestNow() time.Time { + return time.Date(2026, 5, 22, 14, 33, 10, 0, time.UTC) +} diff --git a/doctor/server.go b/doctor/server.go new file mode 100644 index 0000000..e9477d5 --- /dev/null +++ b/doctor/server.go @@ -0,0 +1,167 @@ +package doctor + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "path/filepath" + "sync" + "time" +) + +const ( + doctorRunPath = "/doctor/run" + socketMode = 0o600 + doctorReadHeaderTimeout = 5 * time.Second +) + +type ServerConfig struct { + SocketPath string + Runner Runner +} + +type Server struct { + cfg ServerConfig + + httpServer *http.Server + closeOnce sync.Once + closeErr error + + mu sync.Mutex + running bool +} + +func NewServer(cfg ServerConfig) *Server { + return &Server{cfg: cfg} +} + +func (s *Server) Start(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + if s.cfg.Runner == nil { + return errors.New("doctor runner is required") + } + if s.cfg.SocketPath == "" { + return errors.New("doctor socket path is required") + } + if err := os.MkdirAll(filepath.Dir(s.cfg.SocketPath), 0o750); err != nil { + return fmt.Errorf("create doctor socket directory: %w", err) + } + if err := removeStaleSocket(s.cfg.SocketPath); err != nil { + return err + } + + var listenConfig net.ListenConfig + listener, err := listenConfig.Listen(ctx, "unix", s.cfg.SocketPath) + if err != nil { + return fmt.Errorf("listen on doctor socket: %w", err) + } + chmodErr := os.Chmod(s.cfg.SocketPath, socketMode) + if chmodErr != nil { + _ = listener.Close() + return fmt.Errorf("chmod doctor socket: %w", chmodErr) + } + + mux := http.NewServeMux() + mux.HandleFunc(doctorRunPath, s.handleRun) + s.httpServer = &http.Server{ + Handler: mux, + ReadHeaderTimeout: doctorReadHeaderTimeout, + } + + go func() { + _ = s.httpServer.Serve(listener) + }() + + return nil +} + +func (s *Server) Close(ctx context.Context) error { + s.closeOnce.Do(func() { + if s.httpServer != nil { + if shutdownErr := s.httpServer.Shutdown(ctx); shutdownErr != nil { + _ = s.httpServer.Close() + if !errors.Is(shutdownErr, http.ErrServerClosed) { + s.closeErr = shutdownErr + } + } + } + if removeErr := os.Remove(s.cfg.SocketPath); removeErr != nil { + if !errors.Is(removeErr, os.ErrNotExist) { + s.closeErr = fmt.Errorf("remove doctor socket: %w", removeErr) + } + } + }) + return s.closeErr +} + +func removeStaleSocket(path string) error { + info, err := os.Lstat(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("check stale doctor socket: %w", err) + } + if info.Mode()&os.ModeSocket == 0 { + return fmt.Errorf("refusing to remove non-socket doctor path %s", path) + } + removeErr := os.Remove(path) + if removeErr != nil { + return fmt.Errorf("remove stale doctor socket: %w", removeErr) + } + return nil +} + +func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed\n", http.StatusMethodNotAllowed) + return + } + if !s.beginRun() { + http.Error(w, "doctor scan is already running\n", http.StatusConflict) + return + } + defer s.endRun() + + header := w.Header() + header.Set("Content-Type", "text/plain; charset=utf-8") + + out := flushWriter{w: w} + _, err := s.cfg.Runner.RunDoctor(r.Context(), out) + if err != nil { + _, _ = fmt.Fprintf(out, "\nDoctor failed: %v\n", err) + } +} + +func (s *Server) beginRun() bool { + s.mu.Lock() + defer s.mu.Unlock() + if s.running { + return false + } + s.running = true + return true +} + +func (s *Server) endRun() { + s.mu.Lock() + defer s.mu.Unlock() + s.running = false +} + +type flushWriter struct { + w http.ResponseWriter +} + +func (w flushWriter) Write(p []byte) (int, error) { + n, err := w.w.Write(p) + if flusher, ok := w.w.(http.Flusher); ok { + flusher.Flush() + } + return n, err +} diff --git a/doctor/server_test.go b/doctor/server_test.go new file mode 100644 index 0000000..2097454 --- /dev/null +++ b/doctor/server_test.go @@ -0,0 +1,227 @@ +//go:build !windows + +package doctor + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func TestServerStreamsDoctorOutput(t *testing.T) { + socketPath := shortSocketPath(t) + generatedAt := time.Date(2026, 5, 22, 14, 33, 10, 0, time.UTC) + runner := runnerFunc(func(_ context.Context, out io.Writer) (*Report, error) { + _, _ = io.WriteString(out, "[1/7] Config\n status: ok\n") + return &Report{ + GeneratedAt: generatedAt, + Verdict: VerdictProblemsFound, + ReportPath: "/data/doctor/doctor_2026-05-22T14-33-10Z.json", + }, nil + }) + server := NewServer(ServerConfig{ + SocketPath: socketPath, + Runner: runner, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if err := server.Start(ctx); err != nil { + t.Fatalf("Start() error = %v", err) + } + defer server.Close(context.Background()) + + var out bytes.Buffer + err := RunClient(ctx, socketPath, &out) + if err != nil { + t.Fatalf("RunClient() error = %v", err) + } + + got := out.String() + if !strings.Contains(got, "Connecting to running bundle") { + t.Fatalf("client output = %q, want client connection text", got) + } + if !strings.Contains(got, " socket: "+socketPath) { + t.Fatalf("client output = %q, want socket path", got) + } + if !strings.Contains(got, " status: connected") { + t.Fatalf("client output = %q, want connected status", got) + } + if !strings.Contains(got, "[1/7] Config") { + t.Fatalf("client output = %q, want streamed phase", got) + } +} + +func TestServerAllowsOnlyOneScanAtATime(t *testing.T) { + socketPath := shortSocketPath(t) + started := make(chan struct{}) + release := make(chan struct{}) + var closeStarted sync.Once + runner := runnerFunc(func(ctx context.Context, _ io.Writer) (*Report, error) { + closeStarted.Do(func() { close(started) }) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-release: + return &Report{Verdict: VerdictHealthy}, nil + } + }) + server := NewServer(ServerConfig{ + SocketPath: socketPath, + Runner: runner, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if err := server.Start(ctx); err != nil { + t.Fatalf("Start() error = %v", err) + } + defer server.Close(context.Background()) + + firstDone := make(chan error, 1) + go func() { + var out bytes.Buffer + err := RunClient(ctx, socketPath, &out) + firstDone <- err + }() + + select { + case <-started: + case <-time.After(3 * time.Second): + t.Fatal("first doctor run did not start") + } + + client := unixHTTPClient(socketPath) + response, err := client.Post("http://doctor/doctor/run", "text/plain", nil) + if err != nil { + t.Fatalf("second request error = %v", err) + } + defer response.Body.Close() + + if response.StatusCode != http.StatusConflict { + t.Fatalf("second status = %d, want %d", response.StatusCode, http.StatusConflict) + } + + close(release) + if firstErr := <-firstDone; firstErr != nil { + t.Fatalf("first RunClient() error = %v", firstErr) + } +} + +func TestRunClientReturnsErrorWhenSocketIsMissing(t *testing.T) { + socketPath := filepath.Join(t.TempDir(), "missing.sock") + + var out bytes.Buffer + err := RunClient(context.Background(), socketPath, &out) + + if err == nil { + t.Fatal("RunClient() error = nil, want error") + } +} + +func TestRunClientReturnsNilWhenStartedScanFailsInStream(t *testing.T) { + socketPath := shortSocketPath(t) + runner := runnerFunc(func(_ context.Context, out io.Writer) (*Report, error) { + _, _ = io.WriteString(out, "[1/7] Config\n") + return nil, errors.New("boom") + }) + server := NewServer(ServerConfig{ + SocketPath: socketPath, + Runner: runner, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if err := server.Start(ctx); err != nil { + t.Fatalf("Start() error = %v", err) + } + defer server.Close(context.Background()) + + var out bytes.Buffer + err := RunClient(ctx, socketPath, &out) + if err != nil { + t.Fatalf("RunClient() error = %v, want nil after stream started", err) + } + if !strings.Contains(out.String(), "Doctor failed: boom") { + t.Fatalf("client output does not contain streamed failure:\n%s", out.String()) + } +} + +func TestServerStartRefusesToRemoveRegularFileSocketPath(t *testing.T) { + socketPath := filepath.Join(t.TempDir(), "bundle.sock") + if err := os.WriteFile(socketPath, []byte("not a socket"), 0o600); err != nil { + t.Fatalf("write regular file: %v", err) + } + server := NewServer(ServerConfig{ + SocketPath: socketPath, + Runner: runnerFunc(func(_ context.Context, _ io.Writer) (*Report, error) { + return &Report{Verdict: VerdictHealthy}, nil + }), + }) + + err := server.Start(context.Background()) + if err == nil { + t.Fatal("Start() error = nil, want non-socket refusal") + } + if !strings.Contains(err.Error(), "refusing to remove non-socket") { + t.Fatalf("Start() error = %v", err) + } + raw, readErr := os.ReadFile(socketPath) + if readErr != nil { + t.Fatalf("regular file was removed: %v", readErr) + } + if string(raw) != "not a socket" { + t.Fatalf("regular file content = %q", string(raw)) + } +} + +func TestServerCloseIsIdempotent(t *testing.T) { + socketPath := shortSocketPath(t) + server := NewServer(ServerConfig{ + SocketPath: socketPath, + Runner: runnerFunc(func(_ context.Context, _ io.Writer) (*Report, error) { + return &Report{Verdict: VerdictHealthy}, nil + }), + }) + if err := server.Start(context.Background()); err != nil { + t.Fatalf("Start() error = %v", err) + } + if err := server.Close(context.Background()); err != nil { + t.Fatalf("first Close() error = %v", err) + } + if err := server.Close(context.Background()); err != nil { + t.Fatalf("second Close() error = %v", err) + } + if _, err := os.Stat(socketPath); !os.IsNotExist(err) { + t.Fatalf("socket still exists after Close(): %v", err) + } +} + +func shortSocketPath(t *testing.T) string { + t.Helper() + + // Keep the socket path short enough for Unix socket limits on macOS. + //nolint:usetesting // t.TempDir can be too long for Unix socket paths on macOS. + dir, err := os.MkdirTemp("/tmp", "doctor-test-") + if err != nil { + t.Fatalf("create short temp dir: %v", err) + } + t.Cleanup(func() { + _ = os.RemoveAll(dir) + }) + return filepath.Join(dir, "bundle.sock") +} + +type runnerFunc func(ctx context.Context, out io.Writer) (*Report, error) + +func (f runnerFunc) RunDoctor(ctx context.Context, out io.Writer) (*Report, error) { + return f(ctx, out) +} diff --git a/doctor/types.go b/doctor/types.go new file mode 100644 index 0000000..1681c87 --- /dev/null +++ b/doctor/types.go @@ -0,0 +1,158 @@ +package doctor + +import ( + "context" + "io" + "time" +) + +const currentReportSchema = 1 + +type Verdict string + +const ( + VerdictHealthy Verdict = "bundle_healthy" + VerdictProblemsFound Verdict = "problems_found" + VerdictFilesRequireClientReupload Verdict = "some_files_require_client_reupload" +) + +type Summary struct { + Groups uint64 `json:"groups"` + Spaces uint64 `json:"spaces"` + Files uint64 `json:"files"` + CIDs uint64 `json:"cids"` + Bytes uint64 `json:"bytes"` +} + +type HealthStatus string + +const ( + StatusOK HealthStatus = "ok" + StatusProblem HealthStatus = "problem" + StatusEmpty HealthStatus = "empty" + StatusSkipped HealthStatus = "skipped" +) + +const ( + problemScopeCID = "cid" + problemScopeConfig = "config" + problemScopeFile = "file" + problemScopeGroup = "group" + problemScopeNetwork = "network" + problemScopeRuntime = "runtime" + problemScopeSpace = "space" +) + +type Report struct { + GeneratedAt time.Time `json:"generatedAt"` + Experimental bool `json:"experimental"` + ReportSchema int `json:"reportSchema"` + Build BuildInfo `json:"build,omitempty"` + BundleConfigPath string `json:"bundleConfigPath,omitempty"` + ClientConfigPath string `json:"clientConfigPath,omitempty"` + ReportPath string `json:"reportPath,omitempty"` + Config ConfigReport `json:"config,omitempty"` + Runtime RuntimeReport `json:"runtime,omitempty"` + Network NetworkReport `json:"network,omitempty"` + Verdict Verdict `json:"verdict"` + SuggestedNextAction string `json:"suggestedNextAction,omitempty"` + Summary Summary `json:"summary"` + Inventory Inventory `json:"inventory,omitempty"` + Problems []Problem `json:"problems,omitempty"` +} + +type BuildInfo struct { + Version string `json:"version,omitempty"` + Commit string `json:"commit,omitempty"` + Date string `json:"date,omitempty"` +} + +type ConfigReport struct { + ConfigID string `json:"configId,omitempty"` + NetworkID string `json:"networkId,omitempty"` + PeerID string `json:"peerId,omitempty"` + MongoCoordinatorURI string `json:"mongoCoordinatorUri,omitempty"` + MongoConsensusURI string `json:"mongoConsensusUri,omitempty"` + RedisURI string `json:"redisUri,omitempty"` + S3 *S3Report `json:"s3,omitempty"` + StoragePath string `json:"storagePath,omitempty"` + AdvertisedAddresses []string `json:"advertisedAddresses,omitempty"` +} + +type S3Report struct { + Bucket string `json:"bucket,omitempty"` + Endpoint string `json:"endpoint,omitempty"` + Region string `json:"region,omitempty"` + ForcePathStyle bool `json:"forcePathStyle,omitempty"` +} + +type RuntimeReport struct { + Mongo HealthStatus `json:"mongo,omitempty"` + Redis HealthStatus `json:"redis,omitempty"` + RedisBloom HealthStatus `json:"redisBloom,omitempty"` + Storage HealthStatus `json:"storage,omitempty"` +} + +type NetworkReport struct { + ListenTCPAddr string `json:"listenTcpAddr,omitempty"` + ListenUDPAddr string `json:"listenUdpAddr,omitempty"` + AdvertisedAddresses []string `json:"advertisedAddresses,omitempty"` + TCPListen HealthStatus `json:"tcpListen,omitempty"` + UDPListen HealthStatus `json:"udpListen,omitempty"` + Advertised HealthStatus `json:"advertised,omitempty"` +} + +type Problem struct { + Scope string `json:"scope"` + ID string `json:"id,omitempty"` + Issue string `json:"issue"` + Recoverable string `json:"recoverable,omitempty"` +} + +type Inventory struct { + Groups []GroupReport `json:"groups,omitempty"` + Spaces []SpaceReport `json:"spaces,omitempty"` + Files []FileReport `json:"files,omitempty"` +} + +type GroupReport struct { + ID string `json:"id"` + Spaces uint64 `json:"spaces"` + Files uint64 `json:"files"` + CIDs uint64 `json:"cids"` + Bytes uint64 `json:"bytes"` + Limit uint64 `json:"limit,omitempty"` + AccountLimit uint64 `json:"accountLimit,omitempty"` + IndexProblems uint64 `json:"indexProblems,omitempty"` + Status HealthStatus `json:"status"` +} + +type SpaceReport struct { + ID string `json:"id"` + GroupID string `json:"groupId,omitempty"` + Files uint64 `json:"files"` + CIDs uint64 `json:"cids"` + Bytes uint64 `json:"bytes"` + Limit uint64 `json:"limit,omitempty"` + MissingCIDIndex uint64 `json:"missingCidIndex,omitempty"` + MissingBlocks uint64 `json:"missingBlocks,omitempty"` + CorruptBlocks uint64 `json:"corruptBlocks,omitempty"` + IndexProblems uint64 `json:"indexProblems,omitempty"` + Status HealthStatus `json:"status"` +} + +type FileReport struct { + ID string `json:"id"` + SpaceID string `json:"spaceId"` + GroupID string `json:"groupId,omitempty"` + Size uint64 `json:"size"` + CIDs []string `json:"cids,omitempty"` + CIDSizes map[string]uint64 `json:"-"` + MissingCIDIndex []string `json:"missingCidIndex,omitempty"` + MissingBlocks []string `json:"missingBlocks,omitempty"` + CorruptBlocks []string `json:"corruptBlocks,omitempty"` +} + +type Runner interface { + RunDoctor(ctx context.Context, out io.Writer) (*Report, error) +} diff --git a/flake.lock b/flake.lock index 71ba609..3ca8d6d 100644 --- a/flake.lock +++ b/flake.lock @@ -20,11 +20,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1774106199, - "narHash": "sha256-US5Tda2sKmjrg2lNHQL3jRQ6p96cgfWh3J1QBliQ8Ws=", + "lastModified": 1779508470, + "narHash": "sha256-Ap9KJX+5xHIn3bPIpfNgT6MEXdAECECwo4/rmlQD74M=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "6c9a78c09ff4d6c21d0319114873508a6ec01655", + "rev": "29916453413845e54a65b8a1cf996842300cd299", "type": "github" }, "original": { diff --git a/integration/bundle.go b/integration/bundle.go index 2ba0252..2cc0f9a 100644 --- a/integration/bundle.go +++ b/integration/bundle.go @@ -189,6 +189,27 @@ func (bp *BundleProcess) VerifyPort(port string) error { return nil } +// RunDoctor executes the doctor CLI against the running bundle process. +func (bp *BundleProcess) RunDoctor(ctx context.Context) (string, error) { + binaryPath := filepath.Join(bp.projectRoot, "test-bundle") + cmd := exec.CommandContext(ctx, binaryPath, + "doctor", + "--bundle-config", filepath.Join(bp.tmpDir, "bundle.yml"), + ) + cmd.Dir = bp.projectRoot + + output, err := cmd.CombinedOutput() + if err != nil { + return string(output), fmt.Errorf("doctor failed: %w\n%s", err, output) + } + return string(output), nil +} + +// DoctorReports returns JSON reports written next to the test bundle config. +func (bp *BundleProcess) DoctorReports() ([]string, error) { + return filepath.Glob(filepath.Join(bp.tmpDir, "doctor", "doctor_*.json")) +} + // Stop gracefully stops the bundle process. func (bp *BundleProcess) Stop() error { if bp.cmd.Process == nil { diff --git a/integration/integration_test.go b/integration/integration_test.go index bb18891..57fad8a 100644 --- a/integration/integration_test.go +++ b/integration/integration_test.go @@ -4,6 +4,8 @@ package integration import ( "context" + "os" + "strings" "testing" "time" @@ -45,6 +47,19 @@ func TestBundleFreshInstall(t *testing.T) { require.NoError(t, err, "Port 33010 should be listening") t.Log("Port 33010 is listening") + // Verify doctor can talk to the running process over bundle.sock and + // produce the JSON report a user would attach to an issue. + doctorOutput, err := bundle.RunDoctor(ctx) + require.NoError(t, err, "Doctor should complete against the running bundle") + require.Contains(t, doctorOutput, "[7/7] Report") + require.Contains(t, doctorOutput, "Verdict:") + reports, err := bundle.DoctorReports() + require.NoError(t, err, "Doctor report glob should work") + require.Len(t, reports, 1, "Doctor should write one report") + reportData, err := os.ReadFile(reports[0]) + require.NoError(t, err, "Doctor report should be readable") + require.True(t, strings.Contains(string(reportData), `"verdict"`), "Doctor report should contain verdict") + // Graceful shutdown err = bundle.Stop() require.NoError(t, err, "Bundle should shutdown cleanly")