diff --git a/.agents/skills/mo-self-review/SKILL.md b/.agents/skills/mo-self-review/SKILL.md index 0cbe896eb5b16..2f9ff337c0ad8 100644 --- a/.agents/skills/mo-self-review/SKILL.md +++ b/.agents/skills/mo-self-review/SKILL.md @@ -70,6 +70,7 @@ stream of nitpicks. | Gate | When | Action | |------|------|--------| | **G-SELF-REVIEW** | Before `git push`, before opening/updating a PR, or before declaring a change "done" | Run §1–§4 over the full diff, apply §5 convergence discipline, then check the §7 exit gate. Do not push until it passes. | +| **G-RACE-STRESS** | The diff adds or modifies a Go unit test, or changes behavior directly covered by an existing Go unit test | Run a minimal, explicitly named behavioral set with an adaptive `-race -count=N` budget, then run each owning package completely with `-race -count=1`. Never apply repeated stress to a whole package. A missing real PASS blocks the gate; §6 defines the budget and narrow measurement-test exception. | Scope = the complete diff vs the base branch (`git diff ...HEAD` + staged/unstaged), **not** just the last file you touched. @@ -83,7 +84,7 @@ lens is obvious to another. | Lens | Ask | |------|-----| | **Correctness** | Does each changed function produce the right output for ordinary AND boundary inputs (0, 1, max, empty, nil, overflow)? | -| **Concurrency** | Shared state touched by >1 goroutine? Races, lost wakeups, double-close, ordering assumptions? (`-race` the new tests.) | +| **Concurrency** | Shared state touched by >1 goroutine? Races, lost wakeups, double-close, ordering assumptions? Apply the mandatory race-stress gate in §6. | | **Control path** | Can cancel/close/reject/timeout make progress independently of the blocked operation, or does it wait on the same lock/channel/RPC? | | **State / generation** | Is every transition and failed transition defined? Can old work affect a restarted/reused generation or observe it before admission completes? | | **Resource lifecycle** | Every fd/goroutine/lock/alloc created on the change's paths — closed/released on **every** branch incl. error/panic? (→ §4 Q1) | @@ -172,6 +173,81 @@ ownership graph, wait-for graph, and generation boundary from [references/concurrency-lifecycle.md](references/concurrency-lifecycle.md). Derive the test matrix from semantic axes; do not reuse a remembered case list. +### Mandatory Go unit-test race stress + +Before this review gate passes: + +1. Build a minimal focused set from each newly added or modified `TestXxx` plus + the individual existing regression test(s) that directly prove the changed + behavior or transition. When a shared helper, package/global state, or + background worker changes, choose the representative tests for the affected + contract; the package-wide run in step 5 covers the broader interaction. + If an issue, CI failure, or review comment names a failing `TestXxx`, that + exact test is mandatory in the focused set; adjacent tests are not a + substitute. +2. Prove the selection is non-empty: first enumerate it with `go test -list`, or + verify that the test output names every intended test. A successful command + whose `-run` expression matched nothing is not evidence. +3. Measure each exact test once under `-race`, excluding first-build time, and + choose an adaptive repetition count. Read duration `T` from the test's + terminal event emitted by `go test -json`, not the rounded package summary. + With stress budget `B` and measured test duration `T`, use + `N = clamp(floor(B/T), 1, 100)`. Default `B` to 30 seconds; if `T` is absent, + non-positive, or below timer resolution, use the upper cap `N = 100`. + Adjust `B` for the change's risk and CI budget, and record `T`, `B`, and `N`. + If a pre-fix reproduction has a known occurrence window, override the formula + so the post-fix run covers that window; record why. +4. Run each focused test separately so a slow test does not reduce repetitions + for a fast one: + `go test -race -count=N -run '^TestA$' ./pkg/path`. + Independent commands may run in parallel when they do not contend for the + same external resource. Keep repetitions of one test in the same process so + leaked package/global state remains observable. +5. Then run the entire owning package once under the race detector: + `go test -race -count=1 ./pkg/path`. +6. If the package directly or transitively uses CGo, replace `go test` in all + commands with `.agents/skills/mo-dev/scripts/mo-cgo-test`; follow the + `mo-dev` environment setup. Do not silently skip tests because the local + linker or runtime environment is incomplete. + +Every repeated-stress command must contain an exact `-run` expression naming one +individual test. Never apply adaptive `-count=N` stress to a package pattern or +the repository; full-package race coverage is step 5 and runs only once. + +Use a bounded, test-appropriate `-timeout` when needed. Normal tests, +non-race `-count=N`, coverage runs, or one focused race run do not substitute +for this gate. + +The only routine exception is a measurement-only allocation/performance test +whose oracle is invalidated by race-runtime bookkeeping. Isolate only that +measurement behind `//go:build !race`, keep an equivalent functional test in +the race build, and stress the functional test with the adaptive race budget. +Never hide functional behavior or an ordinary timing assertion behind `!race`. +For any other platform, build-tag, or test-kind constraint, report the exact +test and technical reason; the gate remains blocked until the constraint is +resolved or the reviewer explicitly accepts equivalent validation. + +Before accepting the stress result, audit the test design against recurring MO +flake classes: + +- synchronize phases with channels, callbacks, barriers, or observable + conditions; do not use `time.Sleep` or a tiny deadline as the scheduler; +- assert durable behavior, not a transient map entry, worker ownership, or + which goroutine happened to make progress; +- register cleanup immediately so it runs after failed assertions too; restore + package/global state and stop goroutines, timers, sockets, allocators, and + other caller-owned resources; +- make topology, ordering, IDs, and map-derived choices deterministic; repeated + `-count=N` runs share one test process and must not inherit prior-run state; +- use a generous outer deadline only as a hang guard unless timeout behavior is + itself the contract under test. + +Race success does not prove a timing-, allocation-, or instrumentation-sensitive +oracle under non-race or coverage execution. Run the matching CI mode as +additional evidence when the changed test depends on one of those properties. +All evidence must contain the real exit status and be newer than the final +semantic edit or rebase. + **On a PR (same methodology, later in the lifecycle):** `/code-review ultra ` or `/review `. But the point of *this* skill is to run BEFORE the PR so those find nothing. @@ -190,7 +266,9 @@ skill; for CGo build/test env and MO operator/format specifics, see **mo-dev**. □ state ownership, wait-for dependencies, and generation transitions modeled where applicable □ every finding either FIXED or written to the decision log (§5.2) □ severity calibrated to the merge bar (§5.1) — zero open blockers -□ new/changed tests run green (incl. -race where concurrency changed) +□ every new/modified and directly affected Go behavioral unit test passed focused adaptive -race -count=N, with T/B/N recorded and a proven non-empty selection +□ every !race measurement-only test retains a race-tested functional counterpart +□ every owning package passed completely under -race -count=1 □ test matrix covers every changed transition and evidence is newer than the final edit/rebase □ applicable domain guards passed (index-plugin → §8) — additive to the §1–§4 sweep above, never a substitute for it ``` diff --git a/.agents/skills/mo-self-review/references/concurrency-lifecycle.md b/.agents/skills/mo-self-review/references/concurrency-lifecycle.md index d2b573a13b35e..386f76fd72177 100644 --- a/.agents/skills/mo-self-review/references/concurrency-lifecycle.md +++ b/.agents/skills/mo-self-review/references/concurrency-lifecycle.md @@ -112,8 +112,10 @@ when a remembered list of test names is present. - Assert fail-fast latency with a short outer deadline and verify it did not fire. - Count irreversible side effects and require exactly once, not merely non-zero. - Make test cleanup release blockers even after an assertion fails. -- Run new concurrency tests with `-race`; stress the focused transition set with - `-count=N`, then run the entire owning package under `-race` once. +- Run the minimal, explicitly named set of new, modified, or directly affected + concurrency tests with the adaptive `-race -count=N` budget defined by the + main skill; never apply repeated stress to the package. Then run the entire + owning package under `-race` once. ## 6. Validate The Closure diff --git a/.github/workflows/merge-update-moc.yaml b/.github/workflows/merge-update-moc.yaml deleted file mode 100644 index 8fcf34bf0692c..0000000000000 --- a/.github/workflows/merge-update-moc.yaml +++ /dev/null @@ -1,12 +0,0 @@ -name: Merge Update MOC -on: - pull_request_target: - branches: [ main,'[0-9]+.[0-9]+*' ] - types: - - closed - -jobs: - merge-update-moc: - if: ${{ github.event.pull_request.merged == true }} - uses: matrixorigin/CI/.github/workflows/merge-update-moc.yaml@main - secrets: inherit \ No newline at end of file diff --git a/pkg/sql/plan/function/func_unary.go b/pkg/sql/plan/function/func_unary.go index 678ada9390e83..1f1710f403c41 100644 --- a/pkg/sql/plan/function/func_unary.go +++ b/pkg/sql/plan/function/func_unary.go @@ -7146,6 +7146,8 @@ var userLevelLocks = struct { retainedCloseCleanups map[string]retainedUserLevelLockCloseCleanup cleanupReservations map[detachedUserLevelLockCleanupKey]uint64 retainedCleanupStarted bool + retainedCleanupGen uint64 + retainedCleanupDone chan struct{} }{ counts: make(map[userLevelLockKey]uint64), byOwner: make(map[string]map[string]struct{}), @@ -7991,17 +7993,39 @@ func startRetainedUserLevelLockCleanupWorkerLocked() { return } userLevelLocks.retainedCleanupStarted = true - go runRetainedUserLevelLockCleanupWorker() + generation := userLevelLocks.retainedCleanupGen + done := make(chan struct{}) + userLevelLocks.retainedCleanupDone = done + go runRetainedUserLevelLockCleanupWorker(generation, done) } -func runRetainedUserLevelLockCleanupWorker() { +func retainedUserLevelLockCleanupGenerationActive(generation uint64) bool { + userLevelLocks.Lock() + defer userLevelLocks.Unlock() + return userLevelLocks.retainedCleanupStarted && + userLevelLocks.retainedCleanupGen == generation +} + +func runRetainedUserLevelLockCleanupWorker(generation uint64, done chan struct{}) { + defer close(done) backoff := userLevelLockDetachedCleanupInitialBackoff for { - progress, remaining := runRetainedUserLevelLockCleanupPass() + if !retainedUserLevelLockCleanupGenerationActive(generation) { + return + } + progress, remaining := runRetainedUserLevelLockCleanupPassForGeneration(&generation) + if !retainedUserLevelLockCleanupGenerationActive(generation) { + return + } if !remaining { userLevelLocks.Lock() - if len(userLevelLocks.pendingCleanups) == 0 && len(userLevelLocks.retainedCloseCleanups) == 0 { + if userLevelLocks.retainedCleanupGen == generation && + len(userLevelLocks.pendingCleanups) == 0 && + len(userLevelLocks.retainedCloseCleanups) == 0 { userLevelLocks.retainedCleanupStarted = false + if userLevelLocks.retainedCleanupDone == done { + userLevelLocks.retainedCleanupDone = nil + } userLevelLocks.Unlock() return } @@ -8022,10 +8046,18 @@ func runRetainedUserLevelLockCleanupWorker() { } func runRetainedUserLevelLockCleanupPass() (bool, bool) { + return runRetainedUserLevelLockCleanupPassForGeneration(nil) +} + +func runRetainedUserLevelLockCleanupPassForGeneration(generation *uint64) (bool, bool) { progress := false remaining := false userLevelLocks.Lock() + if generation != nil && userLevelLocks.retainedCleanupGen != *generation { + userLevelLocks.Unlock() + return false, false + } pending := make([]detachedUserLevelLockCleanupRequest, 0, len(userLevelLocks.pendingCleanups)) for _, req := range userLevelLocks.pendingCleanups { pending = append(pending, detachedUserLevelLockCleanupRequest{ @@ -8047,6 +8079,9 @@ func runRetainedUserLevelLockCleanupPass() (bool, bool) { userLevelLocks.Unlock() for _, req := range pending { + if generation != nil && !retainedUserLevelLockCleanupGenerationActive(*generation) { + return progress, false + } attemptCtx, cancel := context.WithTimeout(context.Background(), userLevelLockDetachedCleanupAttemptTimeout) err := unlockUserLevelLockTxnIDs(attemptCtx, req.ls, req.txnIDs) cancel() @@ -8071,6 +8106,9 @@ func runRetainedUserLevelLockCleanupPass() (bool, bool) { } for _, retained := range closeCleanups { + if generation != nil && !retainedUserLevelLockCleanupGenerationActive(*generation) { + return progress, false + } states := userLevelLocksForOwnerSession(retained.owner, retained.sessionID) if len(states) == 0 { userLevelLocks.Lock() diff --git a/pkg/sql/plan/function/func_unary_test.go b/pkg/sql/plan/function/func_unary_test.go index de67c72992a61..6b901d50f002e 100644 --- a/pkg/sql/plan/function/func_unary_test.go +++ b/pkg/sql/plan/function/func_unary_test.go @@ -7286,6 +7286,8 @@ func TestSleep(t *testing.T) { func resetUserLevelLocksForTest(t *testing.T) { t.Helper() userLevelLocks.Lock() + retainedCleanupDone := userLevelLocks.retainedCleanupDone + userLevelLocks.retainedCleanupGen++ userLevelLocks.counts = make(map[userLevelLockKey]uint64) userLevelLocks.byOwner = make(map[string]map[string]struct{}) userLevelLocks.txnIDs = make(map[userLevelLockKey][][]byte) @@ -7294,7 +7296,20 @@ func resetUserLevelLocksForTest(t *testing.T) { userLevelLocks.retainedCloseCleanups = make(map[string]retainedUserLevelLockCloseCleanup) userLevelLocks.cleanupReservations = make(map[detachedUserLevelLockCleanupKey]uint64) userLevelLocks.retainedCleanupStarted = false + userLevelLocks.retainedCleanupDone = nil userLevelLocks.Unlock() + + // A retained worker can already hold a snapshot of the maps cleared above. + // Advancing the generation makes it stop after its current bounded handoff; + // join it before replacing the detached queues so stale work cannot enter the + // next test's generation. + if retainedCleanupDone != nil { + select { + case <-retainedCleanupDone: + case <-time.After(5 * time.Second): + t.Fatal("retained user-level lock cleanup worker did not stop") + } + } resetDetachedUserLevelLockCleanupsForTest() } @@ -7536,12 +7551,12 @@ func (s *userLevelLockTestService) CloseRemoteLockTable(group uint32, tableID, v func runUserLevelLockTest(t *testing.T, fn func([]lockservice.LockService)) { t.Helper() resetUserLevelLocksForTest(t) + defer resetUserLevelLocksForTest(t) state := &userLevelLockTestState{locks: make(map[string]string)} fn([]lockservice.LockService{ &userLevelLockTestService{id: "user-level-lock-1", state: state}, &userLevelLockTestService{id: "user-level-lock-2", state: state}, }) - resetUserLevelLocksForTest(t) } func TestUserLevelLockCleanupTestServiceUnblocksInFlightUnlock(t *testing.T) { @@ -8691,12 +8706,6 @@ func TestReleaseUserLevelLocksOnSessionCloseRetainsSaturatedHandoffAndRecovers(t detachedUserLevelLockCleanups.Unlock() releaseUserLevelLocksOnSessionCloseWithTimeout(holder, 10*time.Millisecond) - require.NotEmpty(t, UserLevelLocksForMigration(holder)) - userLevelLocks.Lock() - _, retained := userLevelLocks.retainedCloseCleanups[owner] - userLevelLocks.Unlock() - require.True(t, retained) - v, err = getUserLevelLock(lockName, 0, contender) require.NoError(t, err) require.Equal(t, int64(0), v) @@ -8721,8 +8730,11 @@ func TestReleaseUserLevelLocksOnSessionCloseRetainsSaturatedHandoffAndRecovers(t } } waitForCleanup: - progress, _ := runRetainedUserLevelLockCleanupPass() - require.True(t, progress) + // The retained cleanup worker starts asynchronously. It may already own + // the cleanup snapshot by the time this goroutine observes the shared + // maps, so assert the durable ownership behavior instead of a transient + // retainedCloseCleanups entry or which goroutine makes progress. + _, _ = runRetainedUserLevelLockCleanupPass() require.Eventually(t, func() bool { return len(UserLevelLocksForMigration(holder)) == 0 }, 3*time.Second, 10*time.Millisecond) @@ -9283,8 +9295,10 @@ func TestTimedOutFailedAttemptCleanupRetainsOwnershipAfterSaturatedHandoff(t *te } timeoutBacklogDrained: detachedUserLevelLockCleanups.Unlock() - progress, _ := runRetainedUserLevelLockCleanupPass() - require.True(t, progress) + // retainDetachedUserLevelLockTxnCleanup starts a worker. Once unlocks + // resume, that worker may finish before this goroutine runs a pass. + // Drive a pass opportunistically, then assert only the terminal state. + _, _ = runRetainedUserLevelLockCleanupPass() require.Eventually(t, func() bool { userLevelLocks.Lock() @@ -9348,8 +9362,10 @@ func TestSuccessfulProbeCleanupRetainsOwnershipAfterSaturatedHandoff(t *testing. } probeBacklogDrained: detachedUserLevelLockCleanups.Unlock() - progress, _ := runRetainedUserLevelLockCleanupPass() - require.True(t, progress) + // The retained worker races this explicit pass after blockUnlock is + // cleared. Either goroutine may complete the cleanup, so progress from + // this particular call is not part of the behavior under test. + _, _ = runRetainedUserLevelLockCleanupPass() require.Eventually(t, func() bool { userLevelLocks.Lock() _, retained := userLevelLocks.pendingCleanups[key] diff --git a/pkg/vm/engine/tae/containers/mock.go b/pkg/vm/engine/tae/containers/mock.go index e200e36f9bc5c..5223ce0fdda61 100644 --- a/pkg/vm/engine/tae/containers/mock.go +++ b/pkg/vm/engine/tae/containers/mock.go @@ -50,6 +50,37 @@ func (p *MockDataProvider) GetColumnProvider(colIdx int) Vector { return p.providers[colIdx] } +func appendMockFloats[T ~float32 | ~float64]( + vec Vector, + rows int, + unique bool, + next func() T, +) { + const maxConsecutiveCollisions = 1024 + + var seen map[T]struct{} + if unique && rows > 0 { + seen = make(map[T]struct{}, rows) + } + consecutiveCollisions := 0 + for appended := 0; appended < rows; { + value := next() + if unique { + if _, ok := seen[value]; ok { + consecutiveCollisions++ + if consecutiveCollisions == maxConsecutiveCollisions { + panic("failed to generate a unique mock float") + } + continue + } + seen[value] = struct{}{} + consecutiveCollisions = 0 + } + vec.Append(value, false) + appended++ + } +} + func MockVector(t types.Type, rows int, unique bool, provider Vector) (vec Vector) { vec = MakeVector(t, common.DefaultAllocator) if provider != nil { @@ -172,17 +203,17 @@ func MockVector(t types.Type, rows int, unique bool, provider Vector) (vec Vecto } } case types.T_float32: - for i := 0; i < rows; i++ { + appendMockFloats(vec, rows, unique, func() float32 { v1 := rand.Intn(math.MaxInt32) v2 := rand.Intn(math.MaxInt32) + 1 - vec.Append(float32(v1)/float32(v2), false) - } + return float32(v1) / float32(v2) + }) case types.T_float64: - for i := 0; i < rows; i++ { + appendMockFloats(vec, rows, unique, func() float64 { v1 := rand.Intn(math.MaxInt32) v2 := rand.Intn(math.MaxInt32) + 1 - vec.Append(float64(v1)/float64(v2), false) - } + return float64(v1) / float64(v2) + }) case types.T_varchar, types.T_char, types.T_binary, types.T_varbinary, types.T_blob, types.T_text: if unique { for i := 0; i < rows; i++ { diff --git a/pkg/vm/engine/tae/containers/mock_test.go b/pkg/vm/engine/tae/containers/mock_test.go new file mode 100644 index 0000000000000..aca7a8d3f414c --- /dev/null +++ b/pkg/vm/engine/tae/containers/mock_test.go @@ -0,0 +1,63 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package containers + +import ( + "testing" + + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common" + "github.com/stretchr/testify/require" +) + +func TestAppendMockFloatsHonorsUnique(t *testing.T) { + t.Run("resample collisions", func(t *testing.T) { + values := []float32{1, 1, 2, 2, 3} + next := 0 + vec := MakeVector(types.T_float32.ToType(), common.DefaultAllocator) + defer vec.Close() + + appendMockFloats(vec, 3, true, func() float32 { + value := values[next] + next++ + return value + }) + + require.Equal(t, 3, vec.Length()) + require.Equal(t, float32(1), vec.Get(0)) + require.Equal(t, float32(2), vec.Get(1)) + require.Equal(t, float32(3), vec.Get(2)) + require.Equal(t, len(values), next) + }) + + t.Run("preserve duplicates when allowed", func(t *testing.T) { + values := []float64{1, 1, 2} + next := 0 + vec := MakeVector(types.T_float64.ToType(), common.DefaultAllocator) + defer vec.Close() + + appendMockFloats(vec, len(values), false, func() float64 { + value := values[next] + next++ + return value + }) + + require.Equal(t, len(values), vec.Length()) + require.Equal(t, float64(1), vec.Get(0)) + require.Equal(t, float64(1), vec.Get(1)) + require.Equal(t, float64(2), vec.Get(2)) + require.Equal(t, len(values), next) + }) +}