1package kv
2
3import (
4 "errors"
5 "io"
6 "os"
7 "path/filepath"
8 "sort"
9 "sync"
10 "syscall"
11
12 "github.com/bnema/vev/pkg/safedir"
13)
14
15// Store is an in-memory key/value map persisted by whole-file atomic rewrite.
16// Set and Delete mutate memory only; Sync is the durability barrier.
17type Store struct {
18 mu sync.Mutex
19 path string
20 lockFile *os.File
21 data map[string][]byte
22 dirty bool
23 closed bool
24}
25
26// Open reads path, or starts empty when it does not exist. Stray .tmp files
27// are ignored. Any invalid existing main file fails closed and is untouched.
28func Open(path string) (*Store, error) {
29 if err := safedir.EnsurePrivate(filepath.Dir(path)); err != nil {
30 return nil, err
31 }
32 lockFile, err := acquireLock(path)
33 if err != nil {
34 return nil, err
35 }
36 fail := func(err error) (*Store, error) {
37 return nil, errors.Join(err, releaseLock(lockFile))
38 }
39
40 raw, err := os.ReadFile(path)
41 if errors.Is(err, os.ErrNotExist) {
42 store := &Store{path: path, lockFile: lockFile, data: make(map[string][]byte), dirty: true}
43 if err := store.syncLocked(); err != nil {
44 return fail(err)
45 }
46 return store, nil
47 }
48 if err != nil {
49 return fail(err)
50 }
51 data, err := decodeFile(raw)
52 if err != nil {
53 return fail(err)
54 }
55 return &Store{path: path, lockFile: lockFile, data: data}, nil
56}
57
58func (s *Store) Get(key []byte) ([]byte, bool) {
59 s.mu.Lock()
60 defer s.mu.Unlock()
61 value, ok := s.data[string(key)]
62 return append([]byte(nil), value...), ok
63}
64
65func (s *Store) Set(key, value []byte) error {
66 s.mu.Lock()
67 defer s.mu.Unlock()
68 if s.closed {
69 return os.ErrClosed
70 }
71 s.data[string(key)] = append([]byte(nil), value...)
72 s.dirty = true
73 return nil
74}
75
76func (s *Store) Delete(key []byte) error {
77 s.mu.Lock()
78 defer s.mu.Unlock()
79 if s.closed {
80 return os.ErrClosed
81 }
82 if _, exists := s.data[string(key)]; exists {
83 delete(s.data, string(key))
84 s.dirty = true
85 }
86 return nil
87}
88
89func (s *Store) Range(fn func(k, v []byte) bool) {
90 s.mu.Lock()
91 keys := make([]string, 0, len(s.data))
92 for key := range s.data {
93 keys = append(keys, key)
94 }
95 sort.Strings(keys)
96 values := make([][]byte, len(keys))
97 for i, key := range keys {
98 values[i] = append([]byte(nil), s.data[key]...)
99 }
100 s.mu.Unlock()
101 for i, key := range keys {
102 if !fn([]byte(key), values[i]) {
103 return
104 }
105 }
106}
107
108// Sync writes path.tmp, fsyncs it, renames it over path, then fsyncs the
109// directory. The main pathname therefore contains either the previous whole
110// map or the next whole map, never a subset of mutations.
111func (s *Store) Sync() error {
112 s.mu.Lock()
113 defer s.mu.Unlock()
114 if s.closed {
115 return os.ErrClosed
116 }
117 return s.syncLocked()
118}
119
120func (s *Store) syncLocked() error {
121 if !s.dirty {
122 return nil
123 }
124 data := encodeFile(s.data)
125 tmp := s.path + ".tmp"
126 file, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
127 if err != nil {
128 return err
129 }
130 if err := writeAll(file, data); err != nil {
131 return errors.Join(err, file.Close())
132 }
133 if err := file.Sync(); err != nil {
134 return errors.Join(err, file.Close())
135 }
136 if err := file.Close(); err != nil {
137 return err
138 }
139 if err := os.Rename(tmp, s.path); err != nil {
140 return err
141 }
142 if err := syncDir(filepath.Dir(s.path)); err != nil {
143 return err
144 }
145 s.dirty = false
146 return nil
147}
148
149// Close syncs buffered mutations before releasing the process lock.
150func (s *Store) Close() error {
151 s.mu.Lock()
152 defer s.mu.Unlock()
153 if s.closed {
154 return nil
155 }
156 err := s.syncLocked()
157 s.closed = true
158 return errors.Join(err, releaseLock(s.lockFile))
159}
160
161// CloseWithoutSync releases the process lock without making buffered mutations
162// durable. It is used when a higher-level transaction rejected those mutations.
163func (s *Store) CloseWithoutSync() error {
164 s.mu.Lock()
165 defer s.mu.Unlock()
166 if s.closed {
167 return nil
168 }
169 s.closed = true
170 return releaseLock(s.lockFile)
171}
172
173func writeAll(file *os.File, data []byte) error {
174 for len(data) > 0 {
175 n, err := file.Write(data)
176 if err != nil {
177 return err
178 }
179 if n == 0 {
180 return io.ErrShortWrite
181 }
182 data = data[n:]
183 }
184 return nil
185}
186
187func acquireLock(path string) (*os.File, error) {
188 file, err := os.OpenFile(path+".lock", os.O_RDWR|os.O_CREATE, 0o600)
189 if err != nil {
190 return nil, err
191 }
192 if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
193 _ = file.Close()
194 return nil, err
195 }
196 return file, nil
197}
198
199func releaseLock(file *os.File) error {
200 if file == nil {
201 return nil
202 }
203 return errors.Join(syscall.Flock(int(file.Fd()), syscall.LOCK_UN), file.Close())
204}
205
206func syncDir(dir string) error {
207 file, err := os.Open(dir)
208 if err != nil {
209 return err
210 }
211 return errors.Join(file.Sync(), file.Close())
212}