vole
Clone
Read only, no credentials:
https://rodent.software/git/vole.git
Write over SSH with a registered key:
git clone ssh://git@0.0.0.0:2222/vole.git
A free and open-source forge that lives in your Git repo.
Materialise the activity feed in an events table
bca9ec24cbbb938f2d349d5f06e2c1f4fca06fcb · Keenan Nemetz · 9d ago · parent 73a732b1
Materialise the activity feed in an events table The previous commit read the feed as a UNION over the source tables. It stopped materialising events in Go, but SQLite still scanned and sorted the union to find a page, so a page cost the history. This replaces it with the design the feed actually wants: an `events` table with one row per event — issue and change opened, comment, approval, state change, release — carrying the time, actor, title and assignees the feed renders. Rows are indexed by (repository, time, id), so a repository page is an index range read; the instance page walks the global time index and filters by repository, and neither sorts history. An unfiltered page total is a stored per-repository count. Maintenance is part of the refresh that already exists: - an entity's events are replaced when the entity is reloaded, which is how a title or assignee edit reaches old events; - state changes are rebuilt into events for the entities the commits name, which also catches a close-and-reopen that leaves the tree where it started and an endpoint diff cannot see; - releases are refreshed from the tags after every push (post-receive) and after a release written in the browser, because a tag is a ref and no metadata commit announces it; - the stored count moves with the rows a refresh adds or removes. Measured with 100,000 events across 50 repositories: a repository page takes 0.3-0.5 ms (page 1 or page 20), an instance page 0.5-0.9 ms, and an unfiltered total 0.1 ms. A filtered total is the one query that scans, at 12 ms for 100,000 events. On the load harness, instance activity rises from 222 to 482 req/s and p50 from 18 ms to 8 ms; activity from 577 to 762 req/s. The repository feed no longer reads tags per request, and nothing reads the metadata branch to render a page.
forge/index/events.go +369
| @@ -0,0 +1,369 @@ | |||
| 1 | package index | ||
| 2 | |||
| 3 | // The activity feed, materialised. | ||
| 4 | // | ||
| 5 | // The feed is a stream of events: an issue opened, a comment, an approval, a | ||
| 6 | // closed or merged change, a release. The old page assembled those from the | ||
| 7 | // source rows on every request and sorted them, so a page cost the repository's | ||
| 8 | // whole history. Here they are rows with their own time and id, indexed by | ||
| 9 | // (repository, time), so a page is an index range read and an unfiltered total | ||
| 10 | // is a stored count. | ||
| 11 | // | ||
| 12 | // The row carries the title and assignees the feed shows rather than joining the | ||
| 13 | // entity at read time: the feed has always shown the current title, and the | ||
| 14 | // entity's events are replaced when the entity changes. URLs are built from ids | ||
| 15 | // and the repository name at render time, so a rename cannot leave a dead link. | ||
| 16 | |||
| 17 | import ( | ||
| 18 | "database/sql" | ||
| 19 | "fmt" | ||
| 20 | "strings" | ||
| 21 | |||
| 22 | "github.com/rodent-software/vole/core/git" | ||
| 23 | "github.com/rodent-software/vole/forge/repo" | ||
| 24 | "github.com/rodent-software/vole/metadata" | ||
| 25 | ) | ||
| 26 | |||
| 27 | // EventRow is one feed event. The renderer turns it into a row: the ids are what | ||
| 28 | // a URL is made of, and the state is what turns a state event into "closed" or | ||
| 29 | // "reopened". | ||
| 30 | type EventRow struct { | ||
| 31 | RepoID string | ||
| 32 | ID string | ||
| 33 | Kind string // issue | change | release | ||
| 34 | Type string // open | comment | approval | state | release | ||
| 35 | At int64 | ||
| 36 | Actor string | ||
| 37 | Title string | ||
| 38 | Assignees string // \x1f-joined, issue events only | ||
| 39 | EntityID string | ||
| 40 | ParentID string | ||
| 41 | CommentID string | ||
| 42 | State string | ||
| 43 | } | ||
| 44 | |||
| 45 | // insertEvent writes one row. | ||
| 46 | func insertEvent(tx execer, repoID string, e EventRow) error { | ||
| 47 | _, err := tx.Exec( | ||
| 48 | `INSERT INTO events(repo_id, id, kind, type, at, actor, title, assignees, | ||
| 49 | entity_id, parent_id, comment_id, state) | ||
| 50 | VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`, | ||
| 51 | repoID, e.ID, e.Kind, e.Type, e.At, e.Actor, e.Title, e.Assignees, | ||
| 52 | e.EntityID, e.ParentID, e.CommentID, e.State) | ||
| 53 | return err | ||
| 54 | } | ||
| 55 | |||
| 56 | // deleteEventsForEntity removes an entity's events and reports how many rows | ||
| 57 | // went, so the repository's stored count can be adjusted by the same amount. | ||
| 58 | func deleteEventsForEntity(tx execer, repoID, kind, id string) (int, error) { | ||
| 59 | res, err := tx.Exec(`DELETE FROM events WHERE repo_id=? AND kind=? AND entity_id=?`, repoID, kind, id) | ||
| 60 | if err != nil { | ||
| 61 | return 0, err | ||
| 62 | } | ||
| 63 | n, err := res.RowsAffected() | ||
| 64 | if err != nil { | ||
| 65 | return 0, err | ||
| 66 | } | ||
| 67 | return int(n), nil | ||
| 68 | } | ||
| 69 | |||
| 70 | // insertIssueEvents replaces one issue's events: the opening, every comment, and | ||
| 71 | // every state change the branch recorded. Deleted comments keep their tombstone | ||
| 72 | // event, as the feed has always shown that something was said. | ||
| 73 | func insertIssueEvents(tx *sql.Tx, repoID string, iss *metadata.Issue) (inserted, deleted int, err error) { | ||
| 74 | deleted, err = deleteEventsForEntity(tx, repoID, "issue", iss.ID) | ||
| 75 | if err != nil { | ||
| 76 | return 0, deleted, err | ||
| 77 | } | ||
| 78 | assignees := strings.Join(iss.Assignees, "\x1f") | ||
| 79 | base := EventRow{ | ||
| 80 | Kind: "issue", At: iss.Created.Unix(), Actor: iss.Author, | ||
| 81 | Title: iss.Title, Assignees: assignees, EntityID: iss.ID, | ||
| 82 | } | ||
| 83 | open := base | ||
| 84 | open.ID = "issue:" + iss.ID + ":open" | ||
| 85 | open.Type = "open" | ||
| 86 | if err := insertEvent(tx, repoID, open); err != nil { | ||
| 87 | return 0, deleted, err | ||
| 88 | } | ||
| 89 | inserted++ | ||
| 90 | for _, c := range iss.Comments { | ||
| 91 | ev := base | ||
| 92 | ev.ID = "issue:" + iss.ID + ":comment:" + c.ID | ||
| 93 | ev.Type = "comment" | ||
| 94 | ev.At = c.Created.Unix() | ||
| 95 | ev.Actor = c.Author | ||
| 96 | ev.ParentID = iss.ID | ||
| 97 | ev.CommentID = c.ID | ||
| 98 | if err := insertEvent(tx, repoID, ev); err != nil { | ||
| 99 | return inserted, deleted, err | ||
| 100 | } | ||
| 101 | inserted++ | ||
| 102 | } | ||
| 103 | states, err := entityStateEvents(tx, repoID, "issue", iss.ID) | ||
| 104 | if err != nil { | ||
| 105 | return inserted, deleted, err | ||
| 106 | } | ||
| 107 | for _, ev := range states { | ||
| 108 | ev.Kind = "issue" | ||
| 109 | ev.Title = iss.Title | ||
| 110 | ev.Assignees = assignees | ||
| 111 | if err := insertEvent(tx, repoID, ev); err != nil { | ||
| 112 | return inserted, deleted, err | ||
| 113 | } | ||
| 114 | inserted++ | ||
| 115 | } | ||
| 116 | return inserted, deleted, nil | ||
| 117 | } | ||
| 118 | |||
| 119 | // insertRequestEvents is the same for a change: opening, comments, approvals and | ||
| 120 | // state changes. | ||
| 121 | func insertRequestEvents(tx *sql.Tx, repoID string, req *metadata.Request) (inserted, deleted int, err error) { | ||
| 122 | deleted, err = deleteEventsForEntity(tx, repoID, "change", req.ID) | ||
| 123 | if err != nil { | ||
| 124 | return 0, deleted, err | ||
| 125 | } | ||
| 126 | base := EventRow{ | ||
| 127 | Kind: "change", At: req.Created.Unix(), Actor: req.Author, | ||
| 128 | Title: req.Title, EntityID: req.ID, | ||
| 129 | } | ||
| 130 | open := base | ||
| 131 | open.ID = "request:" + req.ID + ":open" | ||
| 132 | open.Type = "open" | ||
| 133 | if err := insertEvent(tx, repoID, open); err != nil { | ||
| 134 | return 0, deleted, err | ||
| 135 | } | ||
| 136 | inserted++ | ||
| 137 | for _, c := range req.Comments { | ||
| 138 | ev := base | ||
| 139 | ev.ID = "request:" + req.ID + ":comment:" + c.ID | ||
| 140 | ev.Type = "comment" | ||
| 141 | ev.At = c.Created.Unix() | ||
| 142 | ev.Actor = c.Author | ||
| 143 | ev.ParentID = req.ID | ||
| 144 | ev.CommentID = c.ID | ||
| 145 | if err := insertEvent(tx, repoID, ev); err != nil { | ||
| 146 | return inserted, deleted, err | ||
| 147 | } | ||
| 148 | inserted++ | ||
| 149 | } | ||
| 150 | for _, a := range req.Approvals { | ||
| 151 | ev := base | ||
| 152 | ev.ID = "request:" + req.ID + ":approval:" + a.Principal | ||
| 153 | ev.Type = "approval" | ||
| 154 | ev.At = a.Created.Unix() | ||
| 155 | ev.Actor = a.Principal | ||
| 156 | if err := insertEvent(tx, repoID, ev); err != nil { | ||
| 157 | return inserted, deleted, err | ||
| 158 | } | ||
| 159 | inserted++ | ||
| 160 | } | ||
| 161 | states, err := entityStateEvents(tx, repoID, "request", req.ID) | ||
| 162 | if err != nil { | ||
| 163 | return inserted, deleted, err | ||
| 164 | } | ||
| 165 | for _, ev := range states { | ||
| 166 | ev.Kind = "change" | ||
| 167 | ev.Title = req.Title | ||
| 168 | if err := insertEvent(tx, repoID, ev); err != nil { | ||
| 169 | return inserted, deleted, err | ||
| 170 | } | ||
| 171 | inserted++ | ||
| 172 | } | ||
| 173 | return inserted, deleted, nil | ||
| 174 | } | ||
| 175 | |||
| 176 | // entityStateEvents reads the state changes for one entity. The ordinal is part | ||
| 177 | // of the event id so two changes in the same second keep the branch's order. | ||
| 178 | func entityStateEvents(tx *sql.Tx, repoID, kind, id string) ([]EventRow, error) { | ||
| 179 | rows, err := tx.Query( | ||
| 180 | `SELECT sha, state, actor, at, ordinal FROM state_changes | ||
| 181 | WHERE repo_id=? AND kind=? AND id=? ORDER BY ordinal`, repoID, kind, id) | ||
| 182 | if err != nil { | ||
| 183 | return nil, err | ||
| 184 | } | ||
| 185 | defer rows.Close() | ||
| 186 | var out []EventRow | ||
| 187 | for rows.Next() { | ||
| 188 | var ( | ||
| 189 | sha, state, actor string | ||
| 190 | at int64 | ||
| 191 | ordinal int | ||
| 192 | ) | ||
| 193 | if err := rows.Scan(&sha, &state, &actor, &at, &ordinal); err != nil { | ||
| 194 | return nil, err | ||
| 195 | } | ||
| 196 | prefix := "issue" | ||
| 197 | if kind == "request" { | ||
| 198 | prefix = "request" | ||
| 199 | } | ||
| 200 | out = append(out, EventRow{ | ||
| 201 | ID: fmt.Sprintf("%s:%s:state:%08d", prefix, id, ordinal), Type: "state", | ||
| 202 | At: at, Actor: actor, EntityID: id, State: state, | ||
| 203 | }) | ||
| 204 | } | ||
| 205 | return out, rows.Err() | ||
| 206 | } | ||
| 207 | |||
| 208 | // replaceReleaseEvents replaces a repository's release events with the tags that | ||
| 209 | // exist now. | ||
| 210 | func replaceReleaseEvents(tx *sql.Tx, repoID string, tags []git.Tag) (inserted, deleted int, err error) { | ||
| 211 | res, err := tx.Exec(`DELETE FROM events WHERE repo_id=? AND kind='release'`, repoID) | ||
| 212 | if err != nil { | ||
| 213 | return 0, 0, err | ||
| 214 | } | ||
| 215 | n, err := res.RowsAffected() | ||
| 216 | if err != nil { | ||
| 217 | return 0, 0, err | ||
| 218 | } | ||
| 219 | deleted = int(n) | ||
| 220 | for _, t := range tags { | ||
| 221 | ev := EventRow{ | ||
| 222 | ID: "release:" + t.Name, Kind: "release", Type: "release", | ||
| 223 | At: t.When, Actor: t.Tagger, Title: t.Name, EntityID: t.Name, | ||
| 224 | } | ||
| 225 | if err := insertEvent(tx, repoID, ev); err != nil { | ||
| 226 | return inserted, deleted, err | ||
| 227 | } | ||
| 228 | inserted++ | ||
| 229 | } | ||
| 230 | return inserted, deleted, nil | ||
| 231 | } | ||
| 232 | |||
| 233 | // bumpEventCount keeps the stored total in step with the rows a refresh added or | ||
| 234 | // removed. | ||
| 235 | func bumpEventCount(tx execer, repoID string, delta int) error { | ||
| 236 | if delta == 0 { | ||
| 237 | return nil | ||
| 238 | } | ||
| 239 | _, err := tx.Exec(`UPDATE repos SET events_count = MAX(0, events_count + ?) WHERE id=?`, delta, repoID) | ||
| 240 | return err | ||
| 241 | } | ||
| 242 | |||
| 243 | // RefreshTags replaces a repository's release events from its tags. It is called | ||
| 244 | // after any push (post-receive) and after a release written in the browser, | ||
| 245 | // because a tag is a ref and no metadata write announces it. | ||
| 246 | func (s *Store) RefreshTags(m *repo.Meta) error { | ||
| 247 | tags, err := m.Git().Tags() | ||
| 248 | if err != nil { | ||
| 249 | return err | ||
| 250 | } | ||
| 251 | tx, err := s.db.Begin() | ||
| 252 | if err != nil { | ||
| 253 | return err | ||
| 254 | } | ||
| 255 | defer tx.Rollback() | ||
| 256 | inserted, deleted, err := replaceReleaseEvents(tx, m.ID, tags) | ||
| 257 | if err != nil { | ||
| 258 | return err | ||
| 259 | } | ||
| 260 | if err := bumpEventCount(tx, m.ID, inserted-deleted); err != nil { | ||
| 261 | return err | ||
| 262 | } | ||
| 263 | return tx.Commit() | ||
| 264 | } | ||
| 265 | |||
| 266 | // eventFilter is the WHERE clause the count and the page share. Kind is the | ||
| 267 | // event kind (issue, change, release); the empty kind is the whole feed. | ||
| 268 | func eventFilter(repoIDs []string, kind, actor, assignee string) (string, []any) { | ||
| 269 | conds := []string{`repo_id IN (` + questionMarks(len(repoIDs)) + `)`} | ||
| 270 | args := make([]any, 0, len(repoIDs)+2) | ||
| 271 | for _, id := range repoIDs { | ||
| 272 | args = append(args, id) | ||
| 273 | } | ||
| 274 | if kind != "" { | ||
| 275 | conds = append(conds, `kind = ?`) | ||
| 276 | args = append(args, kind) | ||
| 277 | } | ||
| 278 | if actor != "" { | ||
| 279 | conds = append(conds, `actor = ?`) | ||
| 280 | args = append(args, actor) | ||
| 281 | } | ||
| 282 | if assignee != "" { | ||
| 283 | // Only issue events carry assignees. The column is a \x1f-joined list; | ||
| 284 | // char(31) keeps the delimiter out of the query text. | ||
| 285 | conds = append(conds, | ||
| 286 | `kind = 'issue'`, | ||
| 287 | `instr(char(31)||assignees||char(31), char(31)||?||char(31)) > 0`) | ||
| 288 | args = append(args, assignee) | ||
| 289 | } | ||
| 290 | return strings.Join(conds, " AND "), args | ||
| 291 | } | ||
| 292 | |||
| 293 | // EventCount is how many events the filters select. The unfiltered total is the | ||
| 294 | // stored per-repository count, so a pager does not count a large repository's | ||
| 295 | // history on every page; a filtered total is an index scan without the sort. | ||
| 296 | func (s *Store) EventCount(repoIDs []string, kind, actor, assignee string) (int, error) { | ||
| 297 | if len(repoIDs) == 0 { | ||
| 298 | return 0, nil | ||
| 299 | } | ||
| 300 | if kind == "" && actor == "" && assignee == "" { | ||
| 301 | args := make([]any, 0, len(repoIDs)) | ||
| 302 | for _, id := range repoIDs { | ||
| 303 | args = append(args, id) | ||
| 304 | } | ||
| 305 | var n int | ||
| 306 | err := s.db.QueryRow( | ||
| 307 | `SELECT COALESCE(SUM(events_count),0) FROM repos WHERE id IN (`+questionMarks(len(repoIDs))+`)`, | ||
| 308 | args...).Scan(&n) | ||
| 309 | return n, err | ||
| 310 | } | ||
| 311 | where, args := eventFilter(repoIDs, kind, actor, assignee) | ||
| 312 | var n int | ||
| 313 | err := s.db.QueryRow(`SELECT COUNT(*) FROM events WHERE `+where, args...).Scan(&n) | ||
| 314 | return n, err | ||
| 315 | } | ||
| 316 | |||
| 317 | // eventColumns is the row shape every event query returns. | ||
| 318 | const eventColumns = `repo_id, id, kind, type, at, actor, title, assignees, | ||
| 319 | entity_id, parent_id, comment_id, state` | ||
| 320 | |||
| 321 | // eventPageQuery builds the query for one page. | ||
| 322 | // | ||
| 323 | // One repository is a single range of the (repository, time) index. Several | ||
| 324 | // repositories walk the global time index in order and filter by repository: the | ||
| 325 | // page stops after offset+limit matches, and no sort is involved. Sorting every | ||
| 326 | // visible repository's history to find the page is exactly what this table | ||
| 327 | // exists to avoid; a narrow visible set scans proportionally more index rows, | ||
| 328 | // which is the bounded alternative. | ||
| 329 | func eventPageQuery(repoIDs []string, kind, actor, assignee string, limit, offset int) (string, []any) { | ||
| 330 | where, args := eventFilter(repoIDs, kind, actor, assignee) | ||
| 331 | args = append(args, limit, offset) | ||
| 332 | if len(repoIDs) == 1 { | ||
| 333 | return `SELECT ` + eventColumns + ` FROM events WHERE ` + where + | ||
| 334 | ` ORDER BY at DESC, id DESC LIMIT ? OFFSET ?`, args | ||
| 335 | } | ||
| 336 | return `SELECT ` + eventColumns + ` FROM events INDEXED BY events_time WHERE ` + where + | ||
| 337 | ` ORDER BY at DESC, id DESC LIMIT ? OFFSET ?`, args | ||
| 338 | } | ||
| 339 | |||
| 340 | // Events reads one page of the feed, newest first. | ||
| 341 | func (s *Store) Events(repoIDs []string, kind, actor, assignee string, limit, offset int) ([]EventRow, error) { | ||
| 342 | if len(repoIDs) == 0 || limit <= 0 { | ||
| 343 | return nil, nil | ||
| 344 | } | ||
| 345 | q, args := eventPageQuery(repoIDs, kind, actor, assignee, limit, offset) | ||
| 346 | rows, err := s.db.Query(q, args...) | ||
| 347 | if err != nil { | ||
| 348 | return nil, err | ||
| 349 | } | ||
| 350 | defer rows.Close() | ||
| 351 | var out []EventRow | ||
| 352 | for rows.Next() { | ||
| 353 | var r EventRow | ||
| 354 | if err := rows.Scan(&r.RepoID, &r.ID, &r.Kind, &r.Type, &r.At, &r.Actor, | ||
| 355 | &r.Title, &r.Assignees, &r.EntityID, &r.ParentID, &r.CommentID, &r.State); err != nil { | ||
| 356 | return nil, err | ||
| 357 | } | ||
| 358 | out = append(out, r) | ||
| 359 | } | ||
| 360 | return out, rows.Err() | ||
| 361 | } | ||
| 362 | |||
| 363 | // questionMarks renders "?,?,?" for an IN clause. | ||
| 364 | func questionMarks(n int) string { | ||
| 365 | if n <= 0 { | ||
| 366 | return "" | ||
| 367 | } | ||
| 368 | return strings.TrimSuffix(strings.Repeat("?,", n), ",") | ||
| 369 | } | ||