cmdgc

Documentation

Overview

Package cmdgc implements the "ffs gc" subcommand.

Index

• Variables

Variables

var Command = &command.C{
	Name: "gc",
	Help: `Garbage-collect objects not reachable from known roots.

If no roots are defined, an error is reported without making any changes
unless --force is set. This avoids accidentally deleting everything in a
store without roots.
`,

	SetFlags: command.Flags(flax.MustBind, &gcFlags),

	Run: command.Adapt(func(env *command.Env) error {
		cfg := env.Config.(*config.Settings)
		return cfg.WithStore(env.Context(), func(s filetree.Store) error {
			var keys []string
			for key, err := range s.Roots().List(env.Context(), "") {
				if err != nil {
					return fmt.Errorf("listing roots: %w", err)
				}
				keys = append(keys, key)
			}

			if len(keys) == 0 && !gcFlags.Force {
				return errors.New("there are no root keys defined")
			} else if len(keys) == 0 {
				fmt.Fprint(env, `>> WARNING <<
* No root keys found!
* Proceeding with collection anyway because --force is set

`)
			}

			n, err := s.Files().Len(env.Context())
			if err != nil {
				return err
			} else if n == 0 {
				return errors.New("the store is empty")
			}
			var idxs []*index.Index
			idx := index.New(int(n), &index.Options{FalsePositiveRate: 0.01})
			fmt.Fprintf(env, "Begin GC of %d objects from %d roots\n", n, len(keys))
			dprintf(env, "Roots: %s\n", wrap(keys, 90, "  ", ", "))

			for _, key := range keys {
				rp, err := root.Open(env.Context(), s.Roots(), key)
				if err != nil {
					return fmt.Errorf("opening %q: %w", key, err)
				}
				idx.Add(key)

				if rp.IndexKey != "" {
					rpi, err := config.LoadIndex(env.Context(), s.Files(), rp.IndexKey)
					if err != nil {
						return err
					}
					idxs = append(idxs, rpi)
					idx.Add(rp.IndexKey)
					dprintf(env, "Loaded cached index for %q (%d keys, %s)\n",
						key, rpi.Stats().NumKeys, config.FormatKey(rp.IndexKey),
					)
					continue
				}

				if gcFlags.RequireIndex {
					return fmt.Errorf("missing required index for %q", key)
				}

				rf, err := rp.File(env.Context(), s.Files())
				if err != nil {
					return fmt.Errorf("opening %q: %w", rp.FileKey, err)
				}
				idx.Add(rp.FileKey)

				dprintf(env, "Scanning data reachable from %q (%s)...\n",
					config.PrintableKey(key), config.FormatKey(rp.FileKey))
				scanned := mapset.New[string]()
				start := time.Now()
				if err := rf.Scan(env.Context(), func(si file.ScanItem) bool {
					key := si.Key()
					if scanned.Has(key) {
						return false
					}
					scanned.Add(key)
					idx.Add(key)
					for _, dkey := range si.Data().Keys() {
						idx.Add(dkey)
					}
					return true
				}); err != nil {
					return fmt.Errorf("scanning %q: %w", key, err)
				}
				dprintf(env, "Finished scanning %d objects [%v elapsed]\n",
					idx.Len(), time.Since(start).Truncate(10*time.Millisecond))
			}
			idxs = append(idxs, idx)

			hasKey := func(key string) bool {
				for _, idx := range idxs {
					if idx.Has(key) {
						return true
					}
				}
				return false
			}

			ctx, cancel := context.WithCancelCause(env.Context())
			defer cancel(nil)
			if gcFlags.Limit > 0 {
				t := time.AfterFunc(gcFlags.Limit, func() { cancel(errSweepLimit) })
				defer t.Stop()
				fmt.Fprintf(env, "Begin sweep over %d objects (limit %v)\n", n, gcFlags.Limit)
			} else {
				fmt.Fprintf(env, "Begin sweep over %d objects\n", n)
			}

			g, run := taskgroup.New(cancel).Limit(gcFlags.Tasks)

			start := time.Now()
			var numKeep, numDrop atomic.Int64

			// Sweep phase 1: Collect all the keys eligible for deletion.
			var toDrop mapset.Set[string]
			for key, err := range s.Files().List(ctx, "") {
				if err != nil {
					return err
				}

				if hasKey(key) {
					numKeep.Add(1)
					continue
				}
				toDrop.Add(key)
			}

			if !toDrop.IsEmpty() {
				fmt.Fprintf(env, "Found %d objects to delete\n", toDrop.Len())

				pb := pbar.New(env, int64(toDrop.Len())).Start()
				for key := range toDrop {
					if ctx.Err() != nil {
						break
					}
					run.Go(func() error {
						pb.Add(1)
						err := s.Files().Delete(ctx, key)
						if err == nil || blob.IsKeyNotFound(err) {
							pb.SetMeta(numDrop.Add(1))
							return nil
						} else if !errors.Is(err, context.Canceled) {
							log.Printf("WARNING: delete key %s: %v", config.FormatKey(key), err)
						}
						return err
					})
				}

				serr := g.Wait()
				pb.Stop()
				fmt.Fprintln(env, " *")
				if serr != nil {
					if errors.Is(context.Cause(ctx), errSweepLimit) {
						fmt.Fprintln(env, "(sweep limit reached)")
					} else {
						return fmt.Errorf("sweeping failed: %w", serr)
					}
				}
			}
			fmt.Fprintf(env, "GC complete: keep %d, drop %d [%v elapsed]\n",
				numKeep.Load(), numDrop.Load(), time.Since(start).Truncate(10*time.Millisecond))
			return nil
		})
	}),
}