1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
// Package store provides storage for hakurei instance states, safe for
// concurrent and cross-process use.
package store
import (
"errors"
"io/fs"
"iter"
"os"
"strconv"
"sync"
"syscall"
"hakurei.app/check"
"hakurei.app/hst"
"hakurei.app/internal/lockedfile"
)
// MutexName is the pathname of the file backing [lockedfile.Mutex] of a [Store]
// and [Handle].
const MutexName = "lock"
// A Store keeps track of [hst.State] via a well-known filesystem accessible to
// all hakurei priv-side processes.
//
// Access to store data and related resources are synchronised on a per-segment
// basis via [Handle].
type Store struct {
// Pathname of directory that the store is rooted in.
base *check.Absolute
// All currently known instances of Handle, keyed by their identity.
handles sync.Map
// Inter-process mutex to synchronise operations against the entire store.
//
// Held during List and when initialising previously unknown identities
// during Do. Must not be accessed directly. Callers should use the bigLock
// method instead.
fileMu *lockedfile.Mutex
// For creating the base directory.
mkdirOnce sync.Once
// Stored error value via mkdirOnce.
mkdirErr error
}
// bigLock acquires fileMu on [Store].
// A non-nil error returned by bigLock is of type [hst.AppError].
func (s *Store) bigLock() (unlock func(), err error) {
s.mkdirOnce.Do(func() { s.mkdirErr = os.MkdirAll(s.base.String(), 0700) })
if s.mkdirErr != nil {
return nil, &hst.AppError{
Step: "create state store directory",
Err: s.mkdirErr,
}
}
if unlock, err = s.fileMu.Lock(); err != nil {
return nil, &hst.AppError{
Step: "acquire lock on the state store",
Err: err,
}
}
return
}
// Handle loads or initialises a [Handle] for identity.
// A non-nil error returned by Handle is of type [hst.AppError].
func (s *Store) Handle(identity int) (*Handle, error) {
h := newHandle(s.base, identity)
h.mu.Lock()
if v, ok := s.handles.LoadOrStore(identity, h); ok {
h = v.(*Handle)
} else {
// acquire big lock to initialise previously unknown segment handle
if unlock, err := s.bigLock(); err != nil {
return nil, err
} else {
defer unlock()
}
err := os.MkdirAll(h.Path.String(), 0700)
h.mu.Unlock()
if err != nil && !errors.Is(err, fs.ErrExist) {
// handle methods will likely return ENOENT
s.handles.CompareAndDelete(identity, h)
return nil, &hst.AppError{
Step: "create store segment directory",
Err: err,
}
}
}
return h, nil
}
// SegmentIdentity is produced by the iterator returned by [Store.Segments].
type SegmentIdentity struct {
// Identity of the current segment.
Identity int
// Error encountered while processing this segment.
Err error
}
// Segments returns an iterator over all [SegmentIdentity] known to the [Store].
//
// To obtain a [Handle] on a segment, caller must then call [Store.Handle]. A
// non-nil error returned by segments is of type [hst.AppError].
func (s *Store) Segments() (iter.Seq[SegmentIdentity], int, error) {
// read directory contents, should only contain storeMutexName and identity
var entries []os.DirEntry
// acquire big lock to read store segment list
if unlock, err := s.bigLock(); err != nil {
return nil, -1, err
} else {
entries, err = os.ReadDir(s.base.String())
unlock()
if err != nil && !errors.Is(err, os.ErrNotExist) {
return nil, -1, &hst.AppError{
Step: "read store segments",
Err: err,
}
}
}
// expects lock file
l := len(entries)
if l > 0 {
l--
}
return func(yield func(SegmentIdentity) bool) {
// for error reporting
const step = "process store segment"
for _, ent := range entries {
si := SegmentIdentity{Identity: -1}
// should only be the big lock
if !ent.IsDir() {
if ent.Name() == MutexName {
continue
}
// this should never happen
si.Err = &hst.AppError{
Step: step, Err: syscall.ENOTDIR,
Msg: "skipped non-directory entry " + strconv.Quote(ent.Name()),
}
goto out
}
// failure paths either indicates a serious bug or external interference
if v, err := strconv.Atoi(ent.Name()); err != nil {
si.Err = &hst.AppError{
Step: step, Err: err,
Msg: "skipped non-identity entry " + strconv.Quote(ent.Name()),
}
goto out
} else if v < hst.IdentityStart || v > hst.IdentityEnd {
si.Err = &hst.AppError{
Step: step, Err: syscall.ERANGE,
Msg: "skipped out of bounds entry " + strconv.Itoa(v),
}
goto out
} else {
si.Identity = v
}
out:
if !yield(si) {
break
}
}
}, l, nil
}
// All returns a non-reusable iterator over all [EntryHandle] known to this
// [Store]. The resulting handles may be retained, but are only safe to use
// during the iteration producing them.
//
// Callers must call copyError after completing iteration and handle the error
// accordingly. A non-nil error returned by copyError is of type [hst.AppError].
func (s *Store) All() (entries iter.Seq[*EntryHandle], copyError func() error) {
var savedErr error
return func(yield func(*EntryHandle) bool) {
var segments iter.Seq[SegmentIdentity]
segments, _, savedErr = s.Segments()
if savedErr != nil {
return
}
for si := range segments {
if savedErr = si.Err; savedErr != nil {
return
}
var handle *Handle
if handle, savedErr = s.Handle(si.Identity); savedErr != nil {
return // not reached
}
var unlock func()
if unlock, savedErr = handle.Lock(); savedErr != nil {
return
}
var segmentEntries iter.Seq[*EntryHandle]
if segmentEntries, _, savedErr = handle.Entries(); savedErr != nil {
unlock()
return // not reached: lock has succeeded
}
for eh := range segmentEntries {
if !yield(eh) {
unlock()
return
}
}
unlock()
}
}, func() error { return savedErr }
}
// New returns the address of a new instance of [Store].
//
// Multiple instances of [Store] rooted in the same directory is possible, but
// unsupported.
func New(base *check.Absolute) *Store {
return &Store{
base: base,
fileMu: lockedfile.MutexAt(base.Append(MutexName).String()),
}
}
|