Commit ea2aad2524
Verified · cmc
internal/store/builds.go +48
| @@ -233,6 +233,7 @@ func (s *Store) AppendBuildLog(id int64, chunk []byte) error { | ||
| 233 | 233 | return err |
| 234 | 234 | } |
| 235 | 235 | if n, _ := res.RowsAffected(); n > 0 { |
| 236 | s.wakeBuild(id) | |
| 236 | 237 | return nil |
| 237 | 238 | } |
| 238 | 239 | // Over the cap. The bounds match exactly once: appending the notice puts |
| @@ -241,6 +242,9 @@ func (s *Store) AppendBuildLog(id int64, chunk []byte) error { | ||
| 241 | 242 | UPDATE builds SET log = log || ? |
| 242 | 243 | WHERE id = ? AND length(log) >= ? AND length(log) < ?`, |
| 243 | 244 | truncNotice, id, MaxBuildLog, MaxBuildLog+len(truncNotice)) |
| 245 | if err == nil { | |
| 246 | s.wakeBuild(id) | |
| 247 | } | |
| 244 | 248 | return err |
| 245 | 249 | } |
| 246 | 250 | |
| @@ -255,6 +259,7 @@ func (s *Store) FinishBuild(id int64, status string) error { | ||
| 255 | 259 | if n, _ := res.RowsAffected(); n == 0 { |
| 256 | 260 | return ErrNotFound |
| 257 | 261 | } |
| 262 | s.wakeBuild(id) | |
| 258 | 263 | return nil |
| 259 | 264 | } |
| 260 | 265 | |
| @@ -331,6 +336,48 @@ func (s *Store) BuildLog(id int64) ([]byte, error) { | ||
| 331 | 336 | return log, err |
| 332 | 337 | } |
| 333 | 338 | |
| 339 | // BuildLogWait returns a channel closed by the next append to, finish of | |
| 340 | // or cancel of the build. Take it before reading, so a change between the | |
| 341 | // read and the wait still wakes the reader. Only this process's writes | |
| 342 | // wake it. | |
| 343 | func (s *Store) BuildLogWait(id int64) <-chan struct{} { | |
| 344 | s.logMu.Lock() | |
| 345 | defer s.logMu.Unlock() | |
| 346 | if s.logWait == nil { | |
| 347 | s.logWait = map[int64]chan struct{}{} | |
| 348 | } | |
| 349 | ch, ok := s.logWait[id] | |
| 350 | if !ok { | |
| 351 | ch = make(chan struct{}) | |
| 352 | s.logWait[id] = ch | |
| 353 | } | |
| 354 | return ch | |
| 355 | } | |
| 356 | ||
| 357 | func (s *Store) wakeBuild(id int64) { | |
| 358 | s.logMu.Lock() | |
| 359 | defer s.logMu.Unlock() | |
| 360 | if ch, ok := s.logWait[id]; ok { | |
| 361 | close(ch) | |
| 362 | delete(s.logWait, id) | |
| 363 | } | |
| 364 | } | |
| 365 | ||
| 366 | // BuildLogFrom returns the build's status and its log past offset bytes, | |
| 367 | // read together so a terminal status comes with every byte before it. | |
| 368 | // The cast matters: || stores the log as text, and substr on text counts | |
| 369 | // characters. | |
| 370 | func (s *Store) BuildLogFrom(id, offset int64) (string, []byte, error) { | |
| 371 | var status string | |
| 372 | var chunk []byte | |
| 373 | err := s.DB.QueryRow(`SELECT status, substr(CAST(log AS BLOB), ?) FROM builds WHERE id = ?`, | |
| 374 | offset+1, id).Scan(&status, &chunk) | |
| 375 | if errors.Is(err, sql.ErrNoRows) { | |
| 376 | return "", nil, ErrNotFound | |
| 377 | } | |
| 378 | return status, chunk, err | |
| 379 | } | |
| 380 | ||
| 334 | 381 | // LatestBuild returns the newest build for a repo, optionally narrowed to |
| 335 | 382 | // one job. It is what a status badge reports. |
| 336 | 383 | func (s *Store) LatestBuild(repoID int64, job string) (Build, error) { |
| @@ -399,6 +446,7 @@ func (s *Store) CancelBuild(id int64) error { | ||
| 399 | 446 | if n, _ := res.RowsAffected(); n == 0 { |
| 400 | 447 | return ErrNotFound |
| 401 | 448 | } |
| 449 | s.wakeBuild(id) | |
| 402 | 450 | return nil |
| 403 | 451 | } |
| 404 | 452 | |
internal/store/builds_test.go +101
| @@ -426,3 +426,104 @@ func TestClaimBuildSkipsUntrustedUnlessAsked(t *testing.T) { | ||
| 426 | 426 | t.Fatalf("untrusted claim: err=%v ok=%v number=%d, want %d", err, ok, b.Number, forkBuild) |
| 427 | 427 | } |
| 428 | 428 | } |
| 429 | ||
| 430 | // A follower's channel closes on each kind of change to its build, and | |
| 431 | // only its build. | |
| 432 | func TestBuildLogWaitWakes(t *testing.T) { | |
| 433 | s := open(t) | |
| 434 | if err := s.MigrateUp(); err != nil { | |
| 435 | t.Fatal(err) | |
| 436 | } | |
| 437 | uid, err := s.CreateUser("cmc", true) | |
| 438 | if err != nil { | |
| 439 | t.Fatal(err) | |
| 440 | } | |
| 441 | if _, err := s.CreateRepo("user", uid, "orgo", "public"); err != nil { | |
| 442 | t.Fatal(err) | |
| 443 | } | |
| 444 | newBuild := func() int64 { | |
| 445 | t.Helper() | |
| 446 | id, err := s.CreateBuild(1, "test", "abc123", "main", `["true"]`, "", "", true) | |
| 447 | if err != nil { | |
| 448 | t.Fatal(err) | |
| 449 | } | |
| 450 | return id | |
| 451 | } | |
| 452 | closed := func(ch <-chan struct{}) bool { | |
| 453 | select { | |
| 454 | case <-ch: | |
| 455 | return true | |
| 456 | default: | |
| 457 | return false | |
| 458 | } | |
| 459 | } | |
| 460 | ||
| 461 | a, b := newBuild(), newBuild() | |
| 462 | wa, wb := s.BuildLogWait(a), s.BuildLogWait(b) | |
| 463 | if err := s.AppendBuildLog(a, []byte("x")); err != nil { | |
| 464 | t.Fatal(err) | |
| 465 | } | |
| 466 | if !closed(wa) { | |
| 467 | t.Error("append did not wake its build") | |
| 468 | } | |
| 469 | if closed(wb) { | |
| 470 | t.Error("append woke another build") | |
| 471 | } | |
| 472 | ||
| 473 | mustClaim(t, s, 1) // claims a, the oldest | |
| 474 | wa = s.BuildLogWait(a) | |
| 475 | if err := s.FinishBuild(a, "success"); err != nil { | |
| 476 | t.Fatal(err) | |
| 477 | } | |
| 478 | if !closed(wa) { | |
| 479 | t.Error("finish did not wake") | |
| 480 | } | |
| 481 | ||
| 482 | if err := s.CancelBuild(b); err != nil { | |
| 483 | t.Fatal(err) | |
| 484 | } | |
| 485 | if !closed(wb) { | |
| 486 | t.Error("cancel did not wake") | |
| 487 | } | |
| 488 | } | |
| 489 | ||
| 490 | // Offsets are bytes, not characters: || stores the log as text, and a | |
| 491 | // multibyte character must not shift where the next read starts. | |
| 492 | func TestBuildLogFrom(t *testing.T) { | |
| 493 | s := open(t) | |
| 494 | if err := s.MigrateUp(); err != nil { | |
| 495 | t.Fatal(err) | |
| 496 | } | |
| 497 | uid, err := s.CreateUser("cmc", true) | |
| 498 | if err != nil { | |
| 499 | t.Fatal(err) | |
| 500 | } | |
| 501 | if _, err := s.CreateRepo("user", uid, "orgo", "public"); err != nil { | |
| 502 | t.Fatal(err) | |
| 503 | } | |
| 504 | id, err := s.CreateBuild(1, "test", "abc123", "main", `["true"]`, "", "", true) | |
| 505 | if err != nil { | |
| 506 | t.Fatal(err) | |
| 507 | } | |
| 508 | first := "héllo — ok\n" | |
| 509 | for _, c := range []string{first, "wörld\n"} { | |
| 510 | if err := s.AppendBuildLog(id, []byte(c)); err != nil { | |
| 511 | t.Fatal(err) | |
| 512 | } | |
| 513 | } | |
| 514 | status, all, err := s.BuildLogFrom(id, 0) | |
| 515 | if err != nil || status != "pending" || string(all) != first+"wörld\n" { | |
| 516 | t.Fatalf("from 0: %q %q %v", status, all, err) | |
| 517 | } | |
| 518 | _, rest, err := s.BuildLogFrom(id, int64(len(first))) | |
| 519 | if err != nil || string(rest) != "wörld\n" { | |
| 520 | t.Fatalf("from %d: %q %v", len(first), rest, err) | |
| 521 | } | |
| 522 | _, none, err := s.BuildLogFrom(id, int64(len(all))) | |
| 523 | if err != nil || len(none) != 0 { | |
| 524 | t.Fatalf("from the end: %q %v", none, err) | |
| 525 | } | |
| 526 | if _, _, err := s.BuildLogFrom(9999, 0); err != ErrNotFound { | |
| 527 | t.Fatalf("missing build: %v", err) | |
| 528 | } | |
| 529 | } | |
internal/store/store.go +6
| @@ -12,6 +12,7 @@ import ( | ||
| 12 | 12 | "sort" |
| 13 | 13 | "strconv" |
| 14 | 14 | "strings" |
| 15 | "sync" | |
| 15 | 16 | |
| 16 | 17 | "modernc.org/sqlite" |
| 17 | 18 | ) |
| @@ -21,6 +22,11 @@ var migrationFS embed.FS | ||
| 21 | 22 | |
| 22 | 23 | type Store struct { |
| 23 | 24 | DB *sql.DB |
| 25 | ||
| 26 | // logWait holds one channel per build someone is following, closed | |
| 27 | // by the next change to that build's row (BuildLogWait). | |
| 28 | logMu sync.Mutex | |
| 29 | logWait map[int64]chan struct{} | |
| 24 | 30 | } |
| 25 | 31 | |
| 26 | 32 | // Open opens (creating if needed) the database at path with WAL mode and |