| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | ) |
| 10 | |
| 11 | var ErrRecentSnapshotPreparing = errors.New("session: recent snapshot is preparing") |
| 12 | |
| 13 | type PreparationState string |
| 14 | |
| 15 | const ( |
| 16 | PreparationReady PreparationState = "ready" |
| 17 | PreparationPreparing PreparationState = "preparing" |
| 18 | PreparationFailed PreparationState = "failed" |
| 19 | ) |
| 20 | |
| 21 | type SessionOpenView struct { |
| 22 | Ref SessionRef `json:"session"` |
| 23 | StorageGeneration string `json:"storageGeneration,omitempty"` |
| 24 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 25 | AcceptedSequence uint64 `json:"acceptedSequence"` |
| 26 | DurableSequence uint64 `json:"durableSequence"` |
| 27 | Recent RecentSnapshot `json:"recent"` |
| 28 | Recovery PreparationState `json:"recovery"` |
| 29 | History PreparationState `json:"history"` |
| 30 | Search PreparationState `json:"search"` |
| 31 | CanExecute bool `json:"canExecute"` |
| 32 | } |
| 33 | |
| 34 | type SessionInspection struct { |
| 35 | Ref SessionRef `json:"session"` |
| 36 | StorageGeneration string `json:"storageGeneration,omitempty"` |
| 37 | Commits uint64 `json:"commits"` |
| 38 | Events uint64 `json:"events"` |
| 39 | DurableSequence uint64 `json:"durableSequence"` |
| 40 | } |
| 41 | |
| 42 | // Recent returns the published bounded baseline without opening the history |
| 43 | // locator, search database, or writer lease. |
| 44 | func (q *Query) Recent(ctx context.Context, ref SessionRef) (RecentSnapshot, error) { |
| 45 | if q == nil || q.persistence == nil { |
| 46 | return RecentSnapshot{}, errors.New("session: nil query") |
| 47 | } |
| 48 | if err := ctx.Err(); err != nil { |
| 49 | return RecentSnapshot{}, err |
| 50 | } |
| 51 | if err := ref.validate(q.hostID); err != nil { |
| 52 | return RecentSnapshot{}, err |
| 53 | } |
| 54 | filesystem, ok := q.persistence.(*FilesystemPersistence) |
| 55 | if !ok { |
| 56 | if q.service != nil { |
| 57 | if runtime, live := q.service.Runtime(ref); live { |
| 58 | snapshot := runtime.Session().RecentSnapshot() |
| 59 | q.authorizeRecent(ref, snapshot) |
| 60 | return snapshot, nil |
| 61 | } |
| 62 | } |
| 63 | return RecentSnapshot{}, ErrRecentSnapshotPreparing |
| 64 | } |
| 65 | dir := filepath.Join(filesystem.Root, ref.SessionID) |
| 66 | manifest, err := readStoredManifest(filepath.Join(dir, "manifest.json")) |
| 67 | if err != nil { |
| 68 | return RecentSnapshot{}, err |
| 69 | } |
| 70 | identity, err := readStorageIdentity(dir, manifest) |
| 71 | if err != nil { |
| 72 | if os.IsNotExist(err) || errors.Is(err, ErrStaleGeneration) { |
| 73 | return RecentSnapshot{}, ErrRecentSnapshotPreparing |
| 74 | } |
| 75 | return RecentSnapshot{}, err |
| 76 | } |
| 77 | snapshot, err := readRecentSnapshot(dir, identity) |
| 78 | if os.IsNotExist(err) { |
| 79 | if q.service != nil { |
| 80 | if runtime, live := q.service.Runtime(ref); live { |
| 81 | snapshot := runtime.Session().RecentSnapshot() |
| 82 | q.authorizeRecent(ref, snapshot) |
| 83 | return snapshot, nil |
| 84 | } |
| 85 | } |
| 86 | return RecentSnapshot{}, ErrRecentSnapshotPreparing |
| 87 | } |
| 88 | q.authorizeRecent(ref, snapshot) |
| 89 | return snapshot, err |
| 90 | } |
| 91 | |
| 92 | func (q *Query) authorizeRecent(ref SessionRef, snapshot RecentSnapshot) { |
| 93 | for _, entry := range snapshot.Entries { |
| 94 | if entry.ContentRef != nil && entry.EventSequence <= snapshot.DurableSequence { |
| 95 | q.authorizeContentForGeneration(ref.SessionID, snapshot.StorageGeneration, entry.ContentRef.Digest, entry.ContentRef.Bytes, entry.ContentRef.IndexDigest) |
| 96 | } |
| 97 | } |
| 98 | } |
| 99 | |
| 100 | // OpenSession returns an observation baseline. It never acquires writer |
| 101 | // ownership for a cold session; callers request execution separately. |
| 102 | func (s *Service) OpenSession(ctx context.Context, ref SessionRef) (SessionOpenView, error) { |
| 103 | if s == nil || s.query == nil { |
| 104 | return SessionOpenView{}, errors.New("session: nil service") |
| 105 | } |
| 106 | return s.query.OpenSession(ctx, ref) |
| 107 | } |
| 108 | |
| 109 | func (q *Query) OpenSession(ctx context.Context, ref SessionRef) (SessionOpenView, error) { |
| 110 | if q == nil { |
| 111 | return SessionOpenView{}, errors.New("session: nil query") |
| 112 | } |
| 113 | if err := ref.validate(q.hostID); err != nil { |
| 114 | return SessionOpenView{}, err |
| 115 | } |
| 116 | view := SessionOpenView{Ref: ref, History: PreparationPreparing, Search: PreparationPreparing} |
| 117 | if filesystem, ok := q.persistence.(*FilesystemPersistence); ok { |
| 118 | if info, statErr := os.Stat(historyIndexPath(filesystem.Root, ref.SessionID)); statErr == nil && info.Mode().IsRegular() { |
| 119 | view.History = PreparationReady |
| 120 | } |
| 121 | if info, statErr := os.Stat(searchIndexPath(filesystem.Root, ref.SessionID)); statErr == nil && info.Mode().IsRegular() { |
| 122 | view.Search = PreparationReady |
| 123 | } |
| 124 | } |
| 125 | if q.service != nil { |
| 126 | if runtime, ok := q.service.Runtime(ref); ok { |
| 127 | snapshot, err := q.Recent(ctx, ref) |
| 128 | if err != nil { |
| 129 | return SessionOpenView{}, err |
| 130 | } |
| 131 | view.Recent, view.StorageGeneration = snapshot, snapshot.StorageGeneration |
| 132 | view.SnapshotSequence, view.AcceptedSequence, view.DurableSequence = snapshot.DurableSequence, runtime.Session().EventSequence(), snapshot.DurableSequence |
| 133 | view.Recovery, view.CanExecute = PreparationReady, true |
| 134 | return view, nil |
| 135 | } |
| 136 | } |
| 137 | recent, err := q.Recent(ctx, ref) |
| 138 | if err != nil { |
| 139 | if errors.Is(err, ErrRecentSnapshotPreparing) { |
| 140 | view.Recovery = PreparationPreparing |
| 141 | return view, nil |
| 142 | } |
| 143 | return SessionOpenView{}, err |
| 144 | } |
| 145 | view.Recent, view.StorageGeneration = recent, recent.StorageGeneration |
| 146 | view.SnapshotSequence, view.AcceptedSequence, view.DurableSequence = recent.DurableSequence, recent.DurableSequence, recent.DurableSequence |
| 147 | view.Recovery = PreparationPreparing |
| 148 | return view, nil |
| 149 | } |
| 150 | |
| 151 | // EnsureExecution prepares or reuses the single RuntimeOwner and returns a |
| 152 | // caller-scoped binding. Releasing that binding cannot cancel shared recovery. |
| 153 | func (s *Service) EnsureExecution(ctx context.Context, ref SessionRef) (*ClientBinding, error) { |
| 154 | return s.Open(ctx, ref) |
| 155 | } |
| 156 | |
| 157 | // StreamSession is the explicit complete-history boundary used by export, |
| 158 | // migration and diagnostics. Normal open, paging and search paths never call |
| 159 | // it or construct a cumulative commit slice. |
| 160 | func (q *Query) StreamSession(ctx context.Context, ref SessionRef, visit func(Commit) error) error { |
| 161 | if q == nil || q.persistence == nil || visit == nil { |
| 162 | return errors.New("session: stream requires query and visitor") |
| 163 | } |
| 164 | if err := ref.validate(q.hostID); err != nil { |
| 165 | return err |
| 166 | } |
| 167 | handle, err := q.persistence.Open(ref.SessionID, ReadOnly) |
| 168 | if err != nil { |
| 169 | return err |
| 170 | } |
| 171 | defer handle.Close(context.WithoutCancel(ctx)) |
| 172 | var cursor uint64 |
| 173 | for { |
| 174 | page, err := handle.Read(ctx, cursor, 256) |
| 175 | if err != nil { |
| 176 | return err |
| 177 | } |
| 178 | for _, commit := range page.Commits { |
| 179 | if err := visit(commit); err != nil { |
| 180 | return err |
| 181 | } |
| 182 | } |
| 183 | if !page.Truncated { |
| 184 | return nil |
| 185 | } |
| 186 | if page.Next <= cursor { |
| 187 | return fmt.Errorf("%w: stream cursor did not advance", ErrDamagedStore) |
| 188 | } |
| 189 | cursor = page.Next |
| 190 | } |
| 191 | } |
| 192 | |
| 193 | // InspectSession performs the full durable traversal intentionally omitted |
| 194 | // from OpenSession. A successful result proves every visited transaction and |
| 195 | // externally stored event payload was readable at inspection time. |
| 196 | func (q *Query) InspectSession(ctx context.Context, ref SessionRef) (SessionInspection, error) { |
| 197 | result := SessionInspection{Ref: ref} |
| 198 | if filesystem, ok := q.persistence.(*FilesystemPersistence); ok { |
| 199 | manifest, err := readStoredManifest(filepath.Join(filesystem.Root, ref.SessionID, "manifest.json")) |
| 200 | if err != nil { |
| 201 | return SessionInspection{}, err |
| 202 | } |
| 203 | identity, err := readStorageIdentity(filepath.Join(filesystem.Root, ref.SessionID), manifest) |
| 204 | if err != nil { |
| 205 | return SessionInspection{}, err |
| 206 | } |
| 207 | result.StorageGeneration = identity.Generation |
| 208 | } |
| 209 | err := q.StreamSession(ctx, ref, func(commit Commit) error { |
| 210 | result.Commits++ |
| 211 | result.Events += uint64(len(commit.Events)) |
| 212 | result.DurableSequence = commit.LastSequence() |
| 213 | return nil |
| 214 | }) |
| 215 | return result, err |
| 216 | } |
| 217 |