okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(cli): invoke daemon storage helpersf6da61d

main
branchesmainall refs →
3.1 KiB123 linesraw
1package storage
2
3import (
4	"albedo/cli/internal/daemon"
5	"bytes"
6	"context"
7	"encoding/json"
8	"errors"
9	"fmt"
10	"io"
11	"os/exec"
12	"strings"
13)
14
15// maintenance owns both mutations and the lock. Closing input revokes ownership.
16type maintenance struct {
17	command *exec.Cmd
18	input   io.WriteCloser
19	output  *json.Decoder
20	stderr  bytes.Buffer
21	waited  bool
22}
23
24type maintenanceMessage struct {
25	Result *CleanupResult `json:"result"`
26	Ready  bool           `json:"ready"`
27}
28
29func (s *Service) startMaintenance(ctx context.Context, files []daemon.StorageFile) (*maintenance, error) {
30	command, err := s.command(ctx, "storage", "maintain", s.Home)
31	if err != nil {
32		return nil, err
33	}
34	helper := &maintenance{command: command}
35	helper.command.Stderr = &helper.stderr
36	input, err := helper.command.StdinPipe()
37	if err != nil {
38		return nil, err
39	}
40	helper.input = input
41	output, err := helper.command.StdoutPipe()
42	if err != nil {
43		_ = input.Close()
44		return nil, err
45	}
46	helper.output = json.NewDecoder(output)
47	helper.output.DisallowUnknownFields()
48	if err := helper.command.Start(); err != nil {
49		_ = input.Close()
50		return nil, err
51	}
52	paths := make([]string, 0, len(files))
53	for _, file := range files {
54		paths = append(paths, file.Path)
55	}
56	if err := json.NewEncoder(input).Encode(struct {
57		Paths []string `json:"paths"`
58	}{paths}); err != nil {
59		helper.close()
60		return nil, helper.failure(ctx, err)
61	}
62	var message maintenanceMessage
63	if err := helper.output.Decode(&message); err != nil {
64		helper.close()
65		return nil, helper.failure(ctx, err)
66	}
67	if !message.Ready || message.Result != nil {
68		helper.close()
69		return nil, helper.failure(ctx, errors.New("invalid maintenance readiness"))
70	}
71	return helper, nil
72}
73
74// close always reaps the child, including revalidation failures before authorization.
75func (m *maintenance) close() {
76	_ = m.input.Close()
77	if !m.waited {
78		_ = m.command.Process.Kill()
79		_ = m.command.Wait()
80		m.waited = true
81	}
82}
83
84func (m *maintenance) failure(ctx context.Context, err error) error {
85	if ctx.Err() != nil {
86		return ctx.Err()
87	}
88	return fmt.Errorf("storage maintenance: %w: %s", err, strings.TrimSpace(m.stderr.String()))
89}
90
91func (m *maintenance) apply(ctx context.Context, vacuum bool, fresh daemon.StorageReport) (CleanupResult, error) {
92	var result CleanupResult
93	err := json.NewEncoder(m.input).Encode(struct {
94		Apply     bool  `json:"apply"`
95		Vacuum    bool  `json:"vacuum"`
96		Before    int64 `json:"before"`
97		FreePages int64 `json:"free_pages"`
98	}{true, vacuum, fresh.Database, fresh.DB.FreePages})
99	if err != nil {
100		m.close()
101		return result, m.failure(ctx, err)
102	}
103	var message maintenanceMessage
104	if err = m.output.Decode(&message); err != nil {
105		m.close()
106		return result, m.failure(ctx, err)
107	}
108	if message.Ready || message.Result == nil {
109		m.close()
110		return result, m.failure(ctx, errors.New("invalid maintenance result"))
111	}
112	// Keep input open until the helper finishes, so its EOF watcher cannot
113	// interrupt successful work. A valid result alone does not prove success.
114	err = m.command.Wait()
115	m.waited = true
116	if ctx.Err() != nil {
117		return result, ctx.Err()
118	}
119	if err != nil {
120		return result, m.failure(ctx, err)
121	}
122	return *message.Result, nil
123}