| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strings" |
| 13 | "time" |
| 14 | |
| 15 | "reasonix/internal/fileutil" |
| 16 | ) |
| 17 | |
| 18 | // Fork publishes an independent child session containing the exact durable |
| 19 | // prefix through a completed turn. The cut must also be a logical batch |
| 20 | // boundary, so a child can never inherit half of an atomic operation. |
| 21 | // |
| 22 | // Fork is a Session operation because the inherited prefix is business state: |
| 23 | // the physical layer only writes the child bytes. |
| 24 | // |
| 25 | // A read-only session is a legitimate source. The child is built from the |
| 26 | // parent's durable prefix and no byte is ever written back, so forking a |
| 27 | // session that another process owns, or one read cold from disk, must not |
| 28 | // require the parent's writer lease. |
| 29 | func (s *Session) Fork(ctx context.Context, childDir, childID string, throughSequence uint64) (Manifest, error) { |
| 30 | if s == nil { |
| 31 | return Manifest{}, fmt.Errorf("session: nil parent session") |
| 32 | } |
| 33 | if err := ctx.Err(); err != nil { |
| 34 | return Manifest{}, err |
| 35 | } |
| 36 | if s.WritableHandle() != nil { |
| 37 | if _, err := s.Flush(ctx); err != nil { |
| 38 | return Manifest{}, fmt.Errorf("flush parent before fork: %w", err) |
| 39 | } |
| 40 | } else if !s.cold() { |
| 41 | return Manifest{}, fmt.Errorf("session: fork needs a durable parent log") |
| 42 | } |
| 43 | var prefix []Commit |
| 44 | var cursor uint64 |
| 45 | for { |
| 46 | previous := cursor |
| 47 | page, err := s.Read(ctx, cursor, 1000) |
| 48 | if err != nil { |
| 49 | return Manifest{}, err |
| 50 | } |
| 51 | stop := false |
| 52 | for _, commit := range page.Commits { |
| 53 | if commit.LastSequence() > throughSequence { |
| 54 | stop = true |
| 55 | break |
| 56 | } |
| 57 | prefix = append(prefix, commit) |
| 58 | } |
| 59 | if stop || !page.Truncated { |
| 60 | break |
| 61 | } |
| 62 | if page.Next <= previous { |
| 63 | return Manifest{}, fmt.Errorf("%w: fork cursor did not advance", ErrDamagedStore) |
| 64 | } |
| 65 | cursor = page.Next |
| 66 | } |
| 67 | s.mu.Lock() |
| 68 | parentDir, parentID := s.dirLocked(), s.id |
| 69 | s.mu.Unlock() |
| 70 | if throughSequence > 0 && (len(prefix) == 0 || prefix[len(prefix)-1].LastSequence() != throughSequence) { |
| 71 | return Manifest{}, fmt.Errorf("%w: cut %d", ErrForkBoundaryNotAtomic, throughSequence) |
| 72 | } |
| 73 | return writeForkChild(ctx, parentDir, parentID, prefix, childDir, childID, throughSequence) |
| 74 | } |
| 75 | |
| 76 | // directoryHandle is implemented by physical handles that know their own |
| 77 | // directory. The leased writer and the cold reader both expose it, so a fork |
| 78 | // can locate the parent's owned files without taking the writer lease. |
| 79 | type directoryHandle interface{ Dir() string } |
| 80 | |
| 81 | // cold reports whether this session reads its durable prefix through a cold |
| 82 | // handle rather than a leased writer. |
| 83 | func (s *Session) cold() bool { |
| 84 | if s == nil { |
| 85 | return false |
| 86 | } |
| 87 | s.mu.Lock() |
| 88 | defer s.mu.Unlock() |
| 89 | return s.coldHandle != nil |
| 90 | } |
| 91 | |
| 92 | // dir reports the physical directory backing this session, if any. |
| 93 | func (s *Session) dir() string { |
| 94 | if s == nil { |
| 95 | return "" |
| 96 | } |
| 97 | s.mu.Lock() |
| 98 | defer s.mu.Unlock() |
| 99 | return s.dirLocked() |
| 100 | } |
| 101 | |
| 102 | // dirLocked reports the physical directory backing this session, if any. The |
| 103 | // caller holds s.mu. |
| 104 | func (s *Session) dirLocked() string { |
| 105 | handle := s.coldHandle |
| 106 | if s.binding != nil { |
| 107 | handle = s.binding.handle |
| 108 | } |
| 109 | if directory, ok := handle.(directoryHandle); ok { |
| 110 | return directory.Dir() |
| 111 | } |
| 112 | return "" |
| 113 | } |
| 114 | |
| 115 | // ErrForkActiveAuthority reports a cut whose prefix still carries in-flight |
| 116 | // execution state. The turn ended, but the commit that closed it also opened |
| 117 | // authority a child must not inherit; the caller reports the reason instead of |
| 118 | // trimming the commit. |
| 119 | var ErrForkActiveAuthority = errors.New("session: fork boundary retains active runtime authority") |
| 120 | |
| 121 | // ErrForkBoundaryNotAtomic reports a cut that lands inside a logical commit. |
| 122 | // The caller reports the boundary as unverifiable rather than trimming the |
| 123 | // commit to fit. |
| 124 | var ErrForkBoundaryNotAtomic = errors.New("session: fork cut is not an atomic batch boundary") |
| 125 | |
| 126 | func writeForkChild(ctx context.Context, parentDir, parentID string, prefix []Commit, childDir, childID string, throughSequence uint64) (Manifest, error) { |
| 127 | childDir = filepath.Clean(strings.TrimSpace(childDir)) |
| 128 | childID = strings.TrimSpace(childID) |
| 129 | if childDir == "." || childID == "" { |
| 130 | return Manifest{}, fmt.Errorf("session: child directory and id are required") |
| 131 | } |
| 132 | // filepath.Join("", "attachments") is a relative path, so an empty parent |
| 133 | // would copy unrelated directories instead of the parent's owned files. |
| 134 | if strings.TrimSpace(parentDir) == "" { |
| 135 | return Manifest{}, fmt.Errorf("session: fork requires the parent session directory") |
| 136 | } |
| 137 | parentHeader, hasParentHeader, err := readSessionHeader(parentDir, parentID) |
| 138 | if err != nil { |
| 139 | return Manifest{}, err |
| 140 | } |
| 141 | projection, err := Project(prefix) |
| 142 | if err != nil { |
| 143 | return Manifest{}, err |
| 144 | } |
| 145 | if forkProjectionAvailability(projection, throughSequence) == ForkActiveAuthority { |
| 146 | return Manifest{}, ErrForkActiveAuthority |
| 147 | } |
| 148 | |
| 149 | // Inherited events retain their stable IDs and sequences, while physical |
| 150 | // commit identity and operation id are rebound to the child. Parent |
| 151 | // idempotency keys must never suppress a future child submission. |
| 152 | inherited := make([]Commit, len(prefix)) |
| 153 | for i, original := range prefix { |
| 154 | commit := cloneCommit(original) |
| 155 | commit.Codec = Codec |
| 156 | commit.ID = deterministicID("fork\x00" + childID + "\x00" + original.ID) |
| 157 | commit.OperationID = "inherit:" + parentID + ":" + original.ID |
| 158 | commit.WriterGeneration = 1 |
| 159 | hash, hashErr := hashOperation(childID, commit.TurnID, commit.Events) |
| 160 | if hashErr != nil { |
| 161 | return Manifest{}, hashErr |
| 162 | } |
| 163 | commit.OperationHash = hash |
| 164 | inherited[i] = commit |
| 165 | } |
| 166 | var log bytes.Buffer |
| 167 | if _, err := encodeV4Commits(ctx, &log, contentStoreForSessionDir(childDir), inherited); err != nil { |
| 168 | return Manifest{}, err |
| 169 | } |
| 170 | digest := sha256.Sum256(log.Bytes()) |
| 171 | manifest := Manifest{ |
| 172 | SchemaVersion: SchemaVersion, Codec: Codec, StorageRevision: StorageRevision, ContentRoot: sharedContentRoot, SessionID: childID, CreatedAt: time.Now().UTC(), |
| 173 | InheritedEvents: throughSequence, |
| 174 | Source: &Source{Path: parentDir, Size: int64(log.Len()), SHA256: hex.EncodeToString(digest[:]), Version: Codec}, |
| 175 | } |
| 176 | if _, err := os.Stat(childDir); err == nil { |
| 177 | return Manifest{}, fmt.Errorf("session: child session already exists") |
| 178 | } else if !os.IsNotExist(err) { |
| 179 | return Manifest{}, err |
| 180 | } |
| 181 | parent := filepath.Dir(childDir) |
| 182 | if err := os.MkdirAll(parent, 0o700); err != nil { |
| 183 | return Manifest{}, err |
| 184 | } |
| 185 | tmp, err := os.MkdirTemp(parent, "."+childID+".fork-") |
| 186 | if err != nil { |
| 187 | return Manifest{}, err |
| 188 | } |
| 189 | published := false |
| 190 | defer func() { |
| 191 | if !published { |
| 192 | _ = os.RemoveAll(tmp) |
| 193 | } |
| 194 | }() |
| 195 | if err := writeManifestFile(filepath.Join(tmp, "manifest.json"), manifest); err != nil { |
| 196 | return Manifest{}, err |
| 197 | } |
| 198 | if hasParentHeader { |
| 199 | if err := writeSessionHeader(tmp, SessionHeader{ |
| 200 | SchemaVersion: SessionHeaderSchemaVersion, SessionID: childID, CreatedAt: manifest.CreatedAt, |
| 201 | CWD: parentHeader.CWD, ParentSessionID: parentID, Origin: SessionOriginFork, |
| 202 | }); err != nil { |
| 203 | return Manifest{}, err |
| 204 | } |
| 205 | } |
| 206 | if err := fileutil.AtomicWriteFileStrict(filepath.Join(tmp, currentLogName), log.Bytes(), 0o600); err != nil { |
| 207 | return Manifest{}, err |
| 208 | } |
| 209 | if err := copyOwnedSessionFiles(ctx, parentDir, tmp); err != nil { |
| 210 | return Manifest{}, fmt.Errorf("copy fork attachments: %w", err) |
| 211 | } |
| 212 | if replayed, err := Replay(tmp, nil); err != nil || len(replayed) != len(inherited) { |
| 213 | if err == nil { |
| 214 | err = fmt.Errorf("copied %d of %d commits", len(replayed), len(inherited)) |
| 215 | } |
| 216 | return Manifest{}, fmt.Errorf("validate fork: %w", err) |
| 217 | } |
| 218 | if err := os.Rename(tmp, childDir); err != nil { |
| 219 | return Manifest{}, fmt.Errorf("publish child session: %w", err) |
| 220 | } |
| 221 | published = true |
| 222 | return manifest, nil |
| 223 | } |
| 224 |