| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "sort" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 12 | gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit" |
| 13 | "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 15 | ) |
| 16 | |
| 17 | const maxGitLivePeers = 32 |
| 18 | |
| 19 | // maxGitLiveRoots bounds the conversation roots one live subscription follows. |
| 20 | const maxGitLiveRoots = 100 |
| 21 | |
| 22 | type gitLiveWorker struct { |
| 23 | cancel context.CancelFunc |
| 24 | done chan struct{} |
| 25 | started time.Time |
| 26 | } |
| 27 | type gitLiveTarget struct { |
| 28 | repo gitrelay.Repository |
| 29 | target string |
| 30 | } |
| 31 | |
| 32 | // runGitLive reconciles the current repository snapshot with live workers. A |
| 33 | // worker is removed as soon as its repository, relay, feature or private-peer |
| 34 | // authorization disappears. Rotation gives relays beyond the socket budget a |
| 35 | // turn on every reconciliation pass. |
| 36 | func (t *Tenant) runGitLive(ctx context.Context) { |
| 37 | workers := make(map[string]gitLiveWorker) |
| 38 | var cursor int |
| 39 | defer stopGitLiveWorkers(workers) |
| 40 | for { |
| 41 | if !t.gitLiveEnabled() { |
| 42 | stopGitLiveWorkers(workers) |
| 43 | workers = make(map[string]gitLiveWorker) |
| 44 | } else { |
| 45 | cursor = t.reconcileGitLive(ctx, workers, cursor) |
| 46 | } |
| 47 | timer := time.NewTimer(5 * time.Second) |
| 48 | select { |
| 49 | case <-ctx.Done(): |
| 50 | timer.Stop() |
| 51 | return |
| 52 | case <-timer.C: |
| 53 | } |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | func (t *Tenant) gitLiveEnabled() bool { |
| 58 | p := t.Policy() |
| 59 | return p.Features.Grasp && p.Features.Grasp02 && t.git != nil |
| 60 | } |
| 61 | |
| 62 | func (t *Tenant) reconcileGitLive(ctx context.Context, workers map[string]gitLiveWorker, cursor int) int { |
| 63 | desired := t.gitLiveTargets(ctx) |
| 64 | keys := make([]string, 0, len(desired)) |
| 65 | for key := range desired { |
| 66 | keys = append(keys, key) |
| 67 | } |
| 68 | sort.Strings(keys) |
| 69 | if len(keys) == 0 { |
| 70 | stopGitLiveWorkers(workers) |
| 71 | return 0 |
| 72 | } |
| 73 | if cursor >= len(keys) { |
| 74 | cursor = 0 |
| 75 | } |
| 76 | allowed := gitLiveCohort(keys, cursor) |
| 77 | now := time.Now() |
| 78 | for key, worker := range workers { |
| 79 | _, wanted := desired[key] |
| 80 | _, inCohort := allowed[key] |
| 81 | if !wanted || (!inCohort && now.Sub(worker.started) >= gitLiveDwell) { |
| 82 | worker.cancel() |
| 83 | if worker.done != nil { |
| 84 | <-worker.done |
| 85 | } |
| 86 | delete(workers, key) |
| 87 | } |
| 88 | } |
| 89 | allowedKeys := make([]string, 0, len(allowed)) |
| 90 | for key := range allowed { |
| 91 | allowedKeys = append(allowedKeys, key) |
| 92 | } |
| 93 | sort.Strings(allowedKeys) |
| 94 | for _, key := range allowedKeys { |
| 95 | if len(workers) >= maxGitLivePeers { |
| 96 | break |
| 97 | } |
| 98 | if _, ok := workers[key]; ok { |
| 99 | continue |
| 100 | } |
| 101 | workerCtx, cancel := context.WithCancel(ctx) |
| 102 | done := make(chan struct{}) |
| 103 | workers[key] = gitLiveWorker{cancel: cancel, done: done, started: now} |
| 104 | go func(repo gitrelay.Repository, target string) { |
| 105 | defer close(done) |
| 106 | t.gitLivePeer(workerCtx, repo, target) |
| 107 | }(desired[key].repo, desired[key].target) |
| 108 | } |
| 109 | if len(keys) > maxGitLivePeers { |
| 110 | return (cursor + maxGitLivePeers) % len(keys) |
| 111 | } |
| 112 | return cursor |
| 113 | } |
| 114 | |
| 115 | func gitLiveCohort(keys []string, cursor int) map[string]struct{} { |
| 116 | allowed := make(map[string]struct{}, maxGitLivePeers) |
| 117 | if len(keys) == 0 { |
| 118 | return allowed |
| 119 | } |
| 120 | cursor %= len(keys) |
| 121 | if cursor < 0 { |
| 122 | cursor += len(keys) |
| 123 | } |
| 124 | for i := 0; i < len(keys) && i < maxGitLivePeers; i++ { |
| 125 | allowed[keys[(cursor+i)%len(keys)]] = struct{}{} |
| 126 | } |
| 127 | return allowed |
| 128 | } |
| 129 | |
| 130 | const gitLiveDwell = 2 * time.Minute |
| 131 | |
| 132 | func (t *Tenant) gitLiveTargets(ctx context.Context) map[string]gitLiveTarget { |
| 133 | result := make(map[string]gitLiveTarget) |
| 134 | for _, repo := range t.git.Repositories() { |
| 135 | targets := append([]string(nil), repo.Relays...) |
| 136 | if t.Policy().Features.Grasp03 { |
| 137 | if authors, err := t.gitLiveConversationAuthors(ctx, repo); err == nil { |
| 138 | offset := t.gitOutboxOffset(ctx, repo) |
| 139 | authors = rotateGitAuthors(authors, offset) |
| 140 | if lists, err := t.gitLocalOutboxEvents(ctx, authors); err == nil { |
| 141 | targets = append(targets, gitOutboxRelays(lists, repo.Private, t.Policy().PrivatePeers)...) |
| 142 | } |
| 143 | } |
| 144 | } |
| 145 | for _, target := range targets { |
| 146 | target = normalizeRelayGit(target) |
| 147 | if target == "" || sameRelayGit(target, t.RelayURL(), t.publicURL) || !t.gitLiveAuthorized(repo, target) { |
| 148 | continue |
| 149 | } |
| 150 | key := strings.Join([]string{repo.Owner, repo.Identifier, target, repo.EventID, fmt.Sprint(repo.Private)}, "\x00" ) |
| 151 | result[key] = gitLiveTarget{repo: repo, target: target} |
| 152 | } |
| 153 | } |
| 154 | return result |
| 155 | } |
| 156 | |
| 157 | func (t *Tenant) gitLiveConversationAuthors(ctx context.Context, repo gitrelay.Repository) ([]string, error) { |
| 158 | items := make([]event.Event, 0, 256) |
| 159 | for _, filter := range gitEventFilters(repo)[1:] { |
| 160 | rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 256}) |
| 161 | if err != nil { |
| 162 | return nil, err |
| 163 | } |
| 164 | items = append(items, rows.Events...) |
| 165 | } |
| 166 | return gitConversationAuthors(items, repo.Owner, repo.Maintainers), nil |
| 167 | } |
| 168 | |
| 169 | func rotateGitAuthors(authors []string, offset int) []string { |
| 170 | if len(authors) == 0 { |
| 171 | return nil |
| 172 | } |
| 173 | offset %= len(authors) |
| 174 | if offset < 0 { |
| 175 | offset += len(authors) |
| 176 | } |
| 177 | rotated := append([]string(nil), authors[offset:]...) |
| 178 | return append(rotated, authors[:offset]...) |
| 179 | } |
| 180 | |
| 181 | func (t *Tenant) gitLiveAuthorized(repo gitrelay.Repository, target string) bool { |
| 182 | if !repo.Private && !privatePolicy(t.Policy()) { |
| 183 | return true |
| 184 | } |
| 185 | return privatePeerMatch(target, t.Policy().PrivatePeers) |
| 186 | } |
| 187 | |
| 188 | func stopGitLiveWorkers(workers map[string]gitLiveWorker) { |
| 189 | for key, worker := range workers { |
| 190 | worker.cancel() |
| 191 | if worker.done != nil { |
| 192 | <-worker.done |
| 193 | } |
| 194 | delete(workers, key) |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | // gitLivePeer performs bounded history overlap before opening subscriptions. |
| 199 | // One subscription follows all repository and conversation filters. |
| 200 | func (t *Tenant) gitLivePeer(ctx context.Context, repo gitrelay.Repository, target string) { |
| 201 | backoff := time.Second |
| 202 | for ctx.Err() == nil { |
| 203 | historyCtx, historyCancel := context.WithTimeout(ctx, 20*time.Second) |
| 204 | if err := t.gitSyncPeer(historyCtx, repo, target); err != nil && ctx.Err() == nil { |
| 205 | t.app.telemetry.Logger().Debug("Git live history overlap incomplete" , "target" , target, "error" , err) |
| 206 | } |
| 207 | historyCancel() |
| 208 | conversation, err := t.gitConversationEvents(ctx, repo) |
| 209 | if err != nil { |
| 210 | conversation = nil |
| 211 | } |
| 212 | filters := gitLiveFiltersForRoots(repo, conversation) |
| 213 | readyCtx, readyCancel := context.WithTimeout(ctx, 5*time.Second) |
| 214 | transport, err := t.gitTargetTransport(readyCtx, repo, target) |
| 215 | readyCancel() |
| 216 | if err != nil { |
| 217 | backoff = nextGitLiveBackoff(backoff) |
| 218 | if !waitGitLive(ctx, backoff) { |
| 219 | return |
| 220 | } |
| 221 | continue |
| 222 | } |
| 223 | started := time.Now() |
| 224 | subCtx, subCancel := context.WithTimeout(ctx, 2*time.Minute) |
| 225 | err = transport.SubscribeFilters(subCtx, target, filters, func(item event.Event) error { |
| 226 | t.ingestGitLive(ctx, item) |
| 227 | return nil |
| 228 | }) |
| 229 | subCancel() |
| 230 | if ctx.Err() != nil { |
| 231 | return |
| 232 | } |
| 233 | if err != nil { |
| 234 | if time.Since(started) >= 30*time.Second { |
| 235 | backoff = time.Second |
| 236 | } |
| 237 | backoff = nextGitLiveBackoff(backoff) |
| 238 | if !waitGitLive(ctx, backoff) { |
| 239 | return |
| 240 | } |
| 241 | } |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | func (t *Tenant) ingestGitLive(ctx context.Context, item event.Event) { |
| 246 | err := t.ingest(ctx, item, replication.OriginImport) |
| 247 | if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) { |
| 248 | return |
| 249 | } |
| 250 | } |
| 251 | |
| 252 | func gitLiveFiltersForRoots(repo gitrelay.Repository, conversation []event.Event) []event.Filter { |
| 253 | filters := gitEventFilters(repo) |
| 254 | roots := make([]string, 0, len(conversation)) |
| 255 | for _, item := range conversation { |
| 256 | if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE { |
| 257 | roots = append(roots, item.ID) |
| 258 | } |
| 259 | } |
| 260 | if len(roots) == 0 { |
| 261 | return filters |
| 262 | } |
| 263 | // One subscription carries every filter, so the request stays bounded to |
| 264 | // the newest conversations. Older threads are covered by history passes. |
| 265 | if len(roots) > maxGitLiveRoots { |
| 266 | sorted := append([]event.Event(nil), conversation...) |
| 267 | sortEventsNewestFirst(sorted) |
| 268 | roots = roots[:0] |
| 269 | for _, item := range sorted { |
| 270 | if item.Kind == event.KIND_GIT_PATCH || item.Kind == event.KIND_GIT_PR || item.Kind == event.KIND_GIT_ISSUE { |
| 271 | roots = append(roots, item.ID) |
| 272 | } |
| 273 | if len(roots) == maxGitLiveRoots { |
| 274 | break |
| 275 | } |
| 276 | } |
| 277 | } |
| 278 | kinds := []int{event.KIND_GIT_PR_UPDATE, 1111, 1630, 1631, 1632, 1633} |
| 279 | return append(filters, |
| 280 | event.Filter{Kinds: kinds, Tags: map[string][]string{"e" : roots}}, |
| 281 | event.Filter{Kinds: kinds, Tags: map[string][]string{"E" : roots}}, |
| 282 | ) |
| 283 | } |
| 284 | |
| 285 | func nextGitLiveBackoff(current time.Duration) time.Duration { |
| 286 | if current >= 5*time.Minute { |
| 287 | return 5 * time.Minute |
| 288 | } |
| 289 | return current * 2 |
| 290 | } |
| 291 | |
| 292 | func waitGitLive(ctx context.Context, delay time.Duration) bool { |
| 293 | timer := time.NewTimer(delay) |
| 294 | defer timer.Stop() |
| 295 | select { |
| 296 | case <-ctx.Done(): |
| 297 | return false |
| 298 | case <-timer.C: |
| 299 | return true |
| 300 | } |
| 301 | } |
| 302 | |
| 303 | func liveFilter(repo gitrelay.Repository) event.Filter { |
| 304 | return gitEventFilters(repo)[1] |
| 305 | } |