diff options
| author | Ophestra <cat@gensokyo.uk> | 2025-07-03 04:11:38 +0900 |
|---|---|---|
| committer | Ophestra <cat@gensokyo.uk> | 2025-07-03 04:36:59 +0900 |
| commit | 087959e81bcd52104676ccacedd605e6491b4376 (patch) | |
| tree | 11cd6eccbfd9278f6c99d33191b320bb4e728888 /internal/app/state | |
| parent | e6967b8bbb5ceec3abbd002f52cff1167a969e9e (diff) | |
app: remove split implementation
It is completely nonsensical and highly error-prone to have multiple implementations of this in the same build. This should be switched at compile time instead therefore the split packages are pointless.
Signed-off-by: Ophestra <cat@gensokyo.uk>
Diffstat (limited to 'internal/app/state')
| -rw-r--r-- | internal/app/state/id.go | 48 | ||||
| -rw-r--r-- | internal/app/state/id_test.go | 63 | ||||
| -rw-r--r-- | internal/app/state/join.go | 60 | ||||
| -rw-r--r-- | internal/app/state/multi.go | 372 | ||||
| -rw-r--r-- | internal/app/state/multi_test.go | 9 | ||||
| -rw-r--r-- | internal/app/state/state.go | 49 | ||||
| -rw-r--r-- | internal/app/state/state_test.go | 144 |
7 files changed, 745 insertions, 0 deletions
diff --git a/internal/app/state/id.go b/internal/app/state/id.go new file mode 100644 index 00000000..11bbc3fe --- /dev/null +++ b/internal/app/state/id.go @@ -0,0 +1,48 @@ +package state + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "fmt" +) + +type ID [16]byte + +var ( + ErrInvalidLength = errors.New("string representation must have a length of 32") +) + +func (a *ID) String() string { + return hex.EncodeToString(a[:]) +} + +func NewAppID(id *ID) error { + _, err := rand.Read(id[:]) + return err +} + +func ParseAppID(id *ID, s string) error { + if len(s) != 32 { + return ErrInvalidLength + } + + for i, b := range s { + if b < '0' || b > 'f' { + return fmt.Errorf("invalid char %q at byte %d", b, i) + } + + v := uint8(b) + if v > '9' { + v = 10 + v - 'a' + } else { + v -= '0' + } + if i%2 == 0 { + v <<= 4 + } + id[i/2] += v + } + + return nil +} diff --git a/internal/app/state/id_test.go b/internal/app/state/id_test.go new file mode 100644 index 00000000..abf2c190 --- /dev/null +++ b/internal/app/state/id_test.go @@ -0,0 +1,63 @@ +package state_test + +import ( + "errors" + "testing" + + "hakurei.app/internal/app/state" +) + +func TestParseAppID(t *testing.T) { + t.Run("bad length", func(t *testing.T) { + if err := state.ParseAppID(new(state.ID), "meow"); !errors.Is(err, state.ErrInvalidLength) { + t.Errorf("ParseAppID: error = %v, wantErr = %v", err, state.ErrInvalidLength) + } + }) + + t.Run("bad byte", func(t *testing.T) { + wantErr := "invalid char '\\n' at byte 15" + if err := state.ParseAppID(new(state.ID), "02bc7f8936b2af6\n\ne2535cd71ef0bb7"); err == nil || err.Error() != wantErr { + t.Errorf("ParseAppID: error = %v, wantErr = %v", err, wantErr) + } + }) + + t.Run("fuzz 16 iterations", func(t *testing.T) { + for i := 0; i < 16; i++ { + testParseAppIDWithRandom(t) + } + }) +} + +func FuzzParseAppID(f *testing.F) { + for i := 0; i < 16; i++ { + id := new(state.ID) + if err := state.NewAppID(id); err != nil { + panic(err.Error()) + } + f.Add(id[0], id[1], id[2], id[3], id[4], id[5], id[6], id[7], id[8], id[9], id[10], id[11], id[12], id[13], id[14], id[15]) + } + + f.Fuzz(func(t *testing.T, b0, b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12, b13, b14, b15 byte) { + testParseAppID(t, &state.ID{b0, b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12, b13, b14, b15}) + }) +} + +func testParseAppIDWithRandom(t *testing.T) { + id := new(state.ID) + if err := state.NewAppID(id); err != nil { + t.Fatalf("cannot generate app ID: %v", err) + } + testParseAppID(t, id) +} + +func testParseAppID(t *testing.T, id *state.ID) { + s := id.String() + got := new(state.ID) + if err := state.ParseAppID(got, s); err != nil { + t.Fatalf("cannot parse app ID: %v", err) + } + + if *got != *id { + t.Fatalf("ParseAppID(%#v) = \n%#v, want \n%#v", s, got, id) + } +} diff --git a/internal/app/state/join.go b/internal/app/state/join.go new file mode 100644 index 00000000..2b4011fe --- /dev/null +++ b/internal/app/state/join.go @@ -0,0 +1,60 @@ +package state + +import ( + "errors" + "maps" +) + +var ( + ErrDuplicate = errors.New("store contains duplicates") +) + +/* +Joiner is the interface that wraps the Join method. + +The Join function uses Joiner if available. +*/ +type Joiner interface{ Join() (Entries, error) } + +// Join returns joined state entries of all active aids. +func Join(s Store) (Entries, error) { + if j, ok := s.(Joiner); ok { + return j.Join() + } + + var ( + aids []int + entries = make(Entries) + + el int + res Entries + loadErr error + ) + + if ln, err := s.List(); err != nil { + return nil, err + } else { + aids = ln + } + + for _, aid := range aids { + if _, err := s.Do(aid, func(c Cursor) { + res, loadErr = c.Load() + }); err != nil { + return nil, err + } + + if loadErr != nil { + return nil, loadErr + } + + // save expected length + el = len(entries) + len(res) + maps.Copy(entries, res) + if len(entries) != el { + return nil, ErrDuplicate + } + } + + return entries, nil +} diff --git a/internal/app/state/multi.go b/internal/app/state/multi.go new file mode 100644 index 00000000..80a3abff --- /dev/null +++ b/internal/app/state/multi.go @@ -0,0 +1,372 @@ +package state + +import ( + "encoding/binary" + "encoding/gob" + "errors" + "fmt" + "io" + "io/fs" + "os" + "path" + "strconv" + "sync" + "syscall" + + "hakurei.app/hst" + "hakurei.app/internal/hlog" +) + +// fine-grained locking and access +type multiStore struct { + base string + + // initialised backends + backends *sync.Map + + lock sync.RWMutex +} + +func (s *multiStore) Do(aid int, f func(c Cursor)) (bool, error) { + s.lock.RLock() + defer s.lock.RUnlock() + + // load or initialise new backend + b := new(multiBackend) + b.lock.Lock() + if v, ok := s.backends.LoadOrStore(aid, b); ok { + b = v.(*multiBackend) + } else { + b.path = path.Join(s.base, strconv.Itoa(aid)) + + // ensure directory + if err := os.MkdirAll(b.path, 0700); err != nil && !errors.Is(err, fs.ErrExist) { + s.backends.CompareAndDelete(aid, b) + return false, err + } + + // open locker file + if l, err := os.OpenFile(b.path+".lock", os.O_RDWR|os.O_CREATE, 0600); err != nil { + s.backends.CompareAndDelete(aid, b) + return false, err + } else { + b.lockfile = l + } + b.lock.Unlock() + } + + // lock backend + if err := b.lockFile(); err != nil { + return false, err + } + + // expose backend methods without exporting the pointer + c := new(struct{ *multiBackend }) + c.multiBackend = b + f(b) + // disable access to the backend on a best-effort basis + c.multiBackend = nil + + // unlock backend + return true, b.unlockFile() +} + +func (s *multiStore) List() ([]int, error) { + var entries []os.DirEntry + + // read base directory to get all aids + if v, err := os.ReadDir(s.base); err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, err + } else { + entries = v + } + + aidsBuf := make([]int, 0, len(entries)) + for _, e := range entries { + // skip non-directories + if !e.IsDir() { + hlog.Verbosef("skipped non-directory entry %q", e.Name()) + continue + } + + // skip non-numerical names + if v, err := strconv.Atoi(e.Name()); err != nil { + hlog.Verbosef("skipped non-aid entry %q", e.Name()) + continue + } else { + if v < 0 || v > 9999 { + hlog.Verbosef("skipped out of bounds entry %q", e.Name()) + continue + } + + aidsBuf = append(aidsBuf, v) + } + } + + return append([]int(nil), aidsBuf...), nil +} + +func (s *multiStore) Close() error { + s.lock.Lock() + defer s.lock.Unlock() + + var errs []error + s.backends.Range(func(_, value any) bool { + b := value.(*multiBackend) + errs = append(errs, b.close()) + return true + }) + + return errors.Join(errs...) +} + +type multiBackend struct { + path string + + // created/opened by prepare + lockfile *os.File + + lock sync.RWMutex +} + +func (b *multiBackend) filename(id *ID) string { + return path.Join(b.path, id.String()) +} + +func (b *multiBackend) lockFileAct(lt int) (err error) { + op := "LockAct" + switch lt { + case syscall.LOCK_EX: + op = "Lock" + case syscall.LOCK_UN: + op = "Unlock" + } + + for { + err = syscall.Flock(int(b.lockfile.Fd()), lt) + if !errors.Is(err, syscall.EINTR) { + break + } + } + if err != nil { + return &fs.PathError{ + Op: op, + Path: b.lockfile.Name(), + Err: err, + } + } + return nil +} + +func (b *multiBackend) lockFile() error { + return b.lockFileAct(syscall.LOCK_EX) +} + +func (b *multiBackend) unlockFile() error { + return b.lockFileAct(syscall.LOCK_UN) +} + +// reads all launchers in simpleBackend +// file contents are ignored if decode is false +func (b *multiBackend) load(decode bool) (Entries, error) { + b.lock.RLock() + defer b.lock.RUnlock() + + // read directory contents, should only contain files named after ids + var entries []os.DirEntry + if pl, err := os.ReadDir(b.path); err != nil { + return nil, err + } else { + entries = pl + } + + // allocate as if every entry is valid + // since that should be the case assuming no external interference happens + r := make(Entries, len(entries)) + + for _, e := range entries { + if e.IsDir() { + return nil, fmt.Errorf("unexpected directory %q in store", e.Name()) + } + + id := new(ID) + if err := ParseAppID(id, e.Name()); err != nil { + return nil, err + } + + // run in a function to better handle file closing + if err := func() error { + // open state file for reading + if f, err := os.Open(path.Join(b.path, e.Name())); err != nil { + return err + } else { + defer func() { + if f.Close() != nil { + // unreachable + panic("foreign state file closed prematurely") + } + }() + + s := new(State) + r[*id] = s + + // append regardless, but only parse if required, implements Len + if decode { + if err = b.decodeState(f, s); err != nil { + return err + } + if s.ID != *id { + return fmt.Errorf("state entry %s has unexpected id %s", id, &s.ID) + } + } + + return nil + } + }(); err != nil { + return nil, err + } + } + + return r, nil +} + +// state file consists of an eight byte header, followed by concatenated gobs +// of [hst.Config] and [State], if [State.Config] is not nil or offset < 0, +// the first gob is skipped +func (b *multiBackend) decodeState(r io.ReadSeeker, state *State) error { + offset := make([]byte, 8) + if l, err := r.Read(offset); err != nil { + if errors.Is(err, io.EOF) { + return fmt.Errorf("state file too short: %d bytes", l) + } + return err + } + + // decode volatile state first + var skipConfig bool + { + o := int64(binary.LittleEndian.Uint64(offset)) + skipConfig = o < 0 + + if !skipConfig { + if l, err := r.Seek(o, io.SeekCurrent); err != nil { + return err + } else if l != 8+o { + return fmt.Errorf("invalid seek offset %d", l) + } + } + } + if err := gob.NewDecoder(r).Decode(state); err != nil { + return err + } + + // decode sealed config + if state.Config == nil { + // config must be provided either as part of volatile state, + // or in the config segment + if skipConfig { + return ErrNoConfig + } + + state.Config = new(hst.Config) + if _, err := r.Seek(8, io.SeekStart); err != nil { + return err + } + return gob.NewDecoder(r).Decode(state.Config) + } else { + return nil + } +} + +// Save writes process state to filesystem +func (b *multiBackend) Save(state *State, configWriter io.WriterTo) error { + b.lock.Lock() + defer b.lock.Unlock() + + if configWriter == nil && state.Config == nil { + return ErrNoConfig + } + + statePath := b.filename(&state.ID) + + if f, err := os.OpenFile(statePath, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0600); err != nil { + return err + } else { + defer func() { + if f.Close() != nil { + // unreachable + panic("state file closed prematurely") + } + }() + return b.encodeState(f, state, configWriter) + } +} + +func (b *multiBackend) encodeState(w io.WriteSeeker, state *State, configWriter io.WriterTo) error { + offset := make([]byte, 8) + + // skip header bytes + if _, err := w.Seek(8, io.SeekStart); err != nil { + return err + } + + if configWriter != nil { + // write config gob and encode header + if l, err := configWriter.WriteTo(w); err != nil { + return err + } else { + binary.LittleEndian.PutUint64(offset, uint64(l)) + } + } else { + // offset == -1 indicates absence of config gob + binary.LittleEndian.PutUint64(offset, 0xffffffffffffffff) + } + + // encode volatile state + if err := gob.NewEncoder(w).Encode(state); err != nil { + return err + } + + // write header + if _, err := w.Seek(0, io.SeekStart); err != nil { + return err + } + _, err := w.Write(offset) + return err +} + +func (b *multiBackend) Destroy(id ID) error { + b.lock.Lock() + defer b.lock.Unlock() + + return os.Remove(b.filename(&id)) +} + +func (b *multiBackend) Load() (Entries, error) { + return b.load(true) +} + +func (b *multiBackend) Len() (int, error) { + // rn consists of only nil entries but has the correct length + rn, err := b.load(false) + return len(rn), err +} + +func (b *multiBackend) close() error { + b.lock.Lock() + defer b.lock.Unlock() + + err := b.lockfile.Close() + if err == nil || errors.Is(err, os.ErrInvalid) || errors.Is(err, os.ErrClosed) { + return nil + } + return err +} + +// NewMulti returns an instance of the multi-file store. +func NewMulti(runDir string) Store { + b := new(multiStore) + b.base = path.Join(runDir, "state") + b.backends = new(sync.Map) + return b +} diff --git a/internal/app/state/multi_test.go b/internal/app/state/multi_test.go new file mode 100644 index 00000000..d493a388 --- /dev/null +++ b/internal/app/state/multi_test.go @@ -0,0 +1,9 @@ +package state_test + +import ( + "testing" + + "hakurei.app/internal/app/state" +) + +func TestMulti(t *testing.T) { testStore(t, state.NewMulti(t.TempDir())) } diff --git a/internal/app/state/state.go b/internal/app/state/state.go new file mode 100644 index 00000000..42750ec6 --- /dev/null +++ b/internal/app/state/state.go @@ -0,0 +1,49 @@ +// Package state provides cross-process state tracking for hakurei container instances. +package state + +import ( + "errors" + "io" + "time" + + "hakurei.app/hst" +) + +var ErrNoConfig = errors.New("state does not contain config") + +type Entries map[ID]*State + +type Store interface { + // Do calls f exactly once and ensures store exclusivity until f returns. + // Returns whether f is called and any errors during the locking process. + // Cursor provided to f becomes invalid as soon as f returns. + Do(aid int, f func(c Cursor)) (ok bool, err error) + + // List queries the store and returns a list of aids known to the store. + // Note that some or all returned aids might not have any active apps. + List() (aids []int, err error) + + // Close releases any resources held by Store. + Close() error +} + +// Cursor provides access to the store +type Cursor interface { + Save(state *State, configWriter io.WriterTo) error + Destroy(id ID) error + Load() (Entries, error) + Len() (int, error) +} + +// State is an instance state +type State struct { + // hakurei instance id + ID ID `json:"instance"` + // child process PID value + PID int `json:"pid"` + // sealed app configuration + Config *hst.Config `json:"config"` + + // process start time + Time time.Time `json:"time"` +} diff --git a/internal/app/state/state_test.go b/internal/app/state/state_test.go new file mode 100644 index 00000000..0025918d --- /dev/null +++ b/internal/app/state/state_test.go @@ -0,0 +1,144 @@ +package state_test + +import ( + "bytes" + "encoding/gob" + "io" + "math/rand/v2" + "reflect" + "slices" + "testing" + "time" + + "hakurei.app/hst" + "hakurei.app/internal/app/state" +) + +func testStore(t *testing.T, s state.Store) { + t.Run("list empty store", func(t *testing.T) { + if aids, err := s.List(); err != nil { + t.Fatalf("List: error = %v", err) + } else if len(aids) != 0 { + t.Fatalf("List: aids = %#v", aids) + } + }) + + const ( + insertEntryChecked = iota + insertEntryNoCheck + insertEntryOtherApp + + tl + ) + + var tc [tl]struct { + state state.State + ct bytes.Buffer + } + for i := 0; i < tl; i++ { + makeState(t, &tc[i].state, &tc[i].ct) + } + + do := func(aid int, f func(c state.Cursor)) { + if ok, err := s.Do(aid, f); err != nil { + t.Fatalf("Do: ok = %v, error = %v", ok, err) + } + } + + insert := func(i, aid int) { + do(aid, func(c state.Cursor) { + if err := c.Save(&tc[i].state, &tc[i].ct); err != nil { + t.Fatalf("Save(&tc[%v]): error = %v", i, err) + } + }) + } + + check := func(i, aid int) { + do(aid, func(c state.Cursor) { + if entries, err := c.Load(); err != nil { + t.Fatalf("Load: error = %v", err) + } else if got, ok := entries[tc[i].state.ID]; !ok { + t.Fatalf("Load: entry %s missing", + &tc[i].state.ID) + } else { + got.Time = tc[i].state.Time + tc[i].state.Config = hst.Template() + if !reflect.DeepEqual(got, &tc[i].state) { + t.Fatalf("Load: entry %s got %#v, want %#v", + &tc[i].state.ID, got, &tc[i].state) + } + tc[i].state.Config = nil + } + }) + } + + t.Run("insert entry checked", func(t *testing.T) { + insert(insertEntryChecked, 0) + check(insertEntryChecked, 0) + }) + + t.Run("insert entry unchecked", func(t *testing.T) { + insert(insertEntryNoCheck, 0) + }) + + t.Run("insert entry different aid", func(t *testing.T) { + insert(insertEntryOtherApp, 1) + check(insertEntryOtherApp, 1) + }) + + t.Run("check previous insertion", func(t *testing.T) { + check(insertEntryNoCheck, 0) + }) + + t.Run("list aids", func(t *testing.T) { + if aids, err := s.List(); err != nil { + t.Fatalf("List: error = %v", err) + } else { + slices.Sort(aids) + want := []int{0, 1} + if !slices.Equal(aids, want) { + t.Fatalf("List() = %#v, want %#v", aids, want) + } + } + }) + + t.Run("join store", func(t *testing.T) { + if entries, err := state.Join(s); err != nil { + t.Fatalf("Join: error = %v", err) + } else if len(entries) != 3 { + t.Fatalf("Join(s) = %#v", entries) + } + }) + + t.Run("clear aid 1", func(t *testing.T) { + do(1, func(c state.Cursor) { + if err := c.Destroy(tc[insertEntryOtherApp].state.ID); err != nil { + t.Fatalf("Destroy: error = %v", err) + } + }) + do(1, func(c state.Cursor) { + if l, err := c.Len(); err != nil { + t.Fatalf("Len: error = %v", err) + } else if l != 0 { + t.Fatalf("Len() = %d, want 0", l) + } + }) + }) + + t.Run("close store", func(t *testing.T) { + if err := s.Close(); err != nil { + t.Fatalf("Close: error = %v", err) + } + }) +} + +func makeState(t *testing.T, s *state.State, ct io.Writer) { + if err := state.NewAppID(&s.ID); err != nil { + t.Fatalf("cannot create dummy state: %v", err) + } + if err := gob.NewEncoder(ct).Encode(hst.Template()); err != nil { + t.Fatalf("cannot encode dummy config: %v", err) + } + s.PID = rand.Int() + s.Time = time.Now() +} |
