package main import ( "context" "database/sql" "encoding/json" "flag" "fmt" "log" "os" "os/exec" "regexp" "sort" "strings" "sync/atomic" "time" _ "github.com/go-sql-driver/mysql" connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events" connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound" connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting" "github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate" "github.com/pingcap-inc/nexuscrm-connectors/testing/connectorfixture" ) const ( defaultInstalledPackID = "pack-local-pressure" defaultGeneration = int64(7) normalEnqueueGeneration = int64(0) maxIdentifierLength = 64 seedDBPrefix = "nexuscrm_pressure_" seedDBTimestampLayout = "20060102T150405Z" ) var safeIdentifier = regexp.MustCompile(`^[A-Za-z0-9_]+$`) type runOptions struct { DSN string Out string ReadbackDB string SeedScenario string SeedDB string Scenario string PackID string Generation int64 Explain bool ListScenarios bool BucketSeconds int ObserveSeconds int SummaryOnly bool } type gateResult struct { Gate gateConfig `json:"gate"` Accepted int `json:"accepted"` Materialized int `json:"materialized"` DirtyRowsSeeded int `json:"dirty_rows_seeded,omitempty"` Processed int `json:"processed"` Skipped int `json:"skipped"` Failed int `json:"failed"` Retry int `json:"retry"` Dead int `json:"dead"` Nonterminal int `json:"nonterminal"` ClaimEligibleHeads int `json:"claim_eligible_heads"` CurrentLeaseOwners int `json:"current_lease_owners"` CurrentLeaseTokens int `json:"current_lease_tokens"` CurrentLeasedWork int `json:"current_leased_work_items"` CurrentLeasedLanes int `json:"current_leased_lanes"` MaxLeasedWork int `json:"max_leased_work_items"` MaxLeasedLanes int `json:"max_leased_lanes"` MaxLeasedTokens int `json:"max_leased_tokens"` StatePairs []statePairBucket `json:"state_pairs,omitempty"` Buckets []progressBucket `json:"buckets,omitempty"` LaneOrderingOK bool `json:"lane_ordering_ok"` LaneOrderViolations int `json:"lane_order_violations"` LaneOrderSamples []string `json:"lane_order_violation_samples,omitempty"` ExplainAccessSummary []string `json:"explain_access_summary,omitempty"` ExplainHasBroadScan bool `json:"explain_has_broad_scan"` Error string `json:"error,omitempty"` } type reportSummary struct { ResultCount int `json:"result_count"` Status string `json:"status"` HarnessOnly bool `json:"harness_only"` AcceptanceSurface string `json:"acceptance_surface"` ErrorCount int `json:"error_count"` CompletedCount int `json:"completed_count"` NonterminalCount int `json:"nonterminal_count"` BroadScanCount int `json:"broad_scan_count"` LaneOrderFailureCount int `json:"lane_order_failure_count"` TotalAccepted int `json:"total_accepted"` TotalMaterialized int `json:"total_materialized"` TotalProcessed int `json:"total_processed"` TotalSkipped int `json:"total_skipped"` TotalFailed int `json:"total_failed"` TotalRetry int `json:"total_retry"` TotalDead int `json:"total_dead"` TotalNonterminal int `json:"total_nonterminal"` MaxClaimEligibleHeads int `json:"max_claim_eligible_heads"` MaxLeasedWork int `json:"max_leased_work_items"` MaxLeasedLanes int `json:"max_leased_lanes"` MaxLeasedTokens int `json:"max_leased_tokens"` MaxBucketCount int `json:"max_bucket_count"` Gates []gateSummary `json:"gates,omitempty"` Notes []string `json:"notes,omitempty"` } type gateSummary struct { Name string `json:"name"` Database string `json:"database,omitempty"` Scenario string `json:"scenario,omitempty"` Status string `json:"status"` Accepted int `json:"accepted"` Materialized int `json:"materialized"` Terminal int `json:"terminal"` Nonterminal int `json:"nonterminal"` ClaimEligibleHeads int `json:"claim_eligible_heads"` MaxLeasedWork int `json:"max_leased_work_items"` MaxLeasedLanes int `json:"max_leased_lanes"` MaxLeasedTokens int `json:"max_leased_tokens"` BucketCount int `json:"bucket_count,omitempty"` LastBucketProcessed int `json:"last_bucket_processed,omitempty"` LastBucketSkipped int `json:"last_bucket_skipped,omitempty"` LastBucketFailed int `json:"last_bucket_failed,omitempty"` LastBucketRetry int `json:"last_bucket_retry,omitempty"` LastBucketDead int `json:"last_bucket_dead,omitempty"` LastBucketPending int `json:"last_bucket_nonterminal,omitempty"` LaneOrderingOK bool `json:"lane_ordering_ok"` ExplainHasBroadScan bool `json:"explain_has_broad_scan"` Error string `json:"error,omitempty"` } type catalogSummary struct { ScenarioCount int `json:"scenario_count"` SeedableCount int `json:"seedable_count"` RuntimeRequiredCount int `json:"runtime_required_count"` SeedModes []namedCount `json:"seed_modes,omitempty"` Flags []namedCount `json:"flags,omitempty"` } type namedCount struct { Name string `json:"name"` Count int `json:"count"` } type gateConfig struct { Name string `json:"name"` Database string `json:"database"` Scenario string `json:"scenario,omitempty"` Explain bool `json:"explain,omitempty"` ReadbackDB bool `json:"readback_db,omitempty"` SeedDB bool `json:"seed_db,omitempty"` } type scenarioDefinition struct { Name string `json:"name"` Source string `json:"source"` Events int `json:"events"` Lanes int `json:"lanes"` Workers int `json:"workers"` SeedMode string `json:"seed_mode"` RuntimeRequired bool `json:"runtime_required"` Flags []string `json:"flags,omitempty"` Purpose string `json:"purpose"` } type statePairBucket struct { WorkState string `json:"work_state"` EventStatus string `json:"event_status"` Count int `json:"count"` } type progressBucket struct { ElapsedSeconds float64 `json:"elapsed_seconds"` Processed int `json:"processed"` Skipped int `json:"skipped"` Failed int `json:"failed"` Retry int `json:"retry"` Dead int `json:"dead"` Nonterminal int `json:"nonterminal"` ClaimEligibleHeads int `json:"claim_eligible_heads"` CurrentLeaseOwners int `json:"current_lease_owners"` CurrentLeaseTokens int `json:"current_lease_tokens"` CurrentLeasedWork int `json:"current_leased_work_items"` CurrentLeasedLanes int `json:"current_leased_lanes"` StatePairs []statePairBucket `json:"state_pairs,omitempty"` } type runReport struct { Repo string `json:"repo"` Branch string `json:"branch"` Head string `json:"head"` DSN string `json:"dsn"` GeneratedAt string `json:"generated_at"` DifferenceCoverage []differenceCoverage `json:"local_vs_test1_difference_coverage"` RemainingCoverageGap []string `json:"remaining_local_vs_test1_gaps"` ScenarioCatalog []scenarioDefinition `json:"scenario_catalog,omitempty"` ScenarioSummary catalogSummary `json:"scenario_catalog_summary,omitempty"` Summary reportSummary `json:"summary,omitempty"` Results []gateResult `json:"results"` } type differenceCoverage struct { Difference string `json:"difference"` LocalGate string `json:"local_gate"` Status string `json:"status"` Notes string `json:"notes,omitempty"` } func main() { var opts runOptions flag.StringVar(&opts.DSN, "dsn", "root@tcp(127.0.0.1:4000)/?parseTime=true&loc=UTC&multiStatements=true", "TiDB/MySQL DSN") flag.StringVar(&opts.Out, "out", "", "optional JSON output path") flag.StringVar(&opts.ReadbackDB, "readback-db", "", "existing pressure database to read without seeding or running workers") flag.StringVar(&opts.SeedScenario, "seed-scenario", "", "metadata catalog scenario to seed into a new local database without running workers") flag.StringVar(&opts.SeedDB, "seed-db", "", "optional new database name for -seed-scenario; defaults to a generated nexuscrm_pressure_* name") flag.StringVar(&opts.Scenario, "scenario", "", "optional scenario catalog name to annotate readback output") flag.StringVar(&opts.PackID, "pack-id", defaultInstalledPackID, "InstalledPackId to read") flag.Int64Var(&opts.Generation, "generation", defaultGeneration, "WorkItem generation to read") flag.BoolVar(&opts.Explain, "explain", false, "include EXPLAIN summary for the claimable WorkItem read path") flag.BoolVar(&opts.ListScenarios, "list-scenarios", false, "print the metadata-only scenario catalog and exit without opening the database") flag.IntVar(&opts.BucketSeconds, "progress-bucket-seconds", 0, "when positive, sample readback counters every N seconds") flag.IntVar(&opts.ObserveSeconds, "observe-seconds", 0, "total seconds to observe progress buckets; requires -progress-bucket-seconds") flag.BoolVar(&opts.SummaryOnly, "summary-only", false, "write only report metadata and aggregate result summary; omit detailed result rows") flag.Parse() if opts.BucketSeconds < 0 || opts.ObserveSeconds < 0 { log.Fatal("progress bucket and observe durations must be non-negative") } if (opts.BucketSeconds > 0) != (opts.ObserveSeconds > 0) { log.Fatal("use -progress-bucket-seconds and -observe-seconds together") } if opts.ListScenarios { report := baseReport(opts) report.DifferenceCoverage = append(report.DifferenceCoverage, scenarioCatalogCoverage()) report.RemainingCoverageGap = scenarioCatalogGaps() report.ScenarioCatalog = scenarioCatalog() report.ScenarioSummary = summarizeCatalog(report.ScenarioCatalog) report.Summary = summarizeReport(nil) if opts.SummaryOnly { report = compactSummaryReport(report) } if err := writeReport(opts.Out, report); err != nil { log.Fatal(err) } return } if strings.TrimSpace(opts.Scenario) != "" { if _, ok := findScenario(opts.Scenario); !ok { log.Fatalf("unknown -scenario %q; known scenarios: %s", opts.Scenario, strings.Join(scenarioNames(), ", ")) } } if strings.TrimSpace(opts.SeedScenario) != "" && strings.TrimSpace(opts.ReadbackDB) != "" { log.Fatal("use either -seed-scenario or -readback-db, not both") } if strings.TrimSpace(opts.SeedScenario) == "" && strings.TrimSpace(opts.ReadbackDB) == "" { log.Fatal("missing -readback-db or -seed-scenario") } ctx := context.Background() db, err := sql.Open("mysql", opts.DSN) if err != nil { log.Fatal(err) } defer db.Close() if err := db.PingContext(ctx); err != nil { log.Fatal(err) } result := gateResult{} if strings.TrimSpace(opts.SeedScenario) != "" { result = seedScenario(ctx, db, &opts) } else { result = readback(ctx, db, opts) } report := baseReport(opts) report.DifferenceCoverage = []differenceCoverage{ { Difference: "Existing local database readback", LocalGate: "readback", Status: "covered", Notes: "Reports state counts, event/work materialization, WorkItem row-lease width, and optional claim SELECT access path without mutating the database.", }, { Difference: "Direct WorkItem schema and seed helper", LocalGate: "seed_scenario", Status: "covered", Notes: "Creates a new local database for direct WorkItem catalog scenarios, seeds rows, and immediately readbacks counts without starting workers.", }, { Difference: "Lane-order violation sample capture for local readback and seed-only scenarios", LocalGate: "readback_lane_order_samples", Status: "covered", Notes: "Reports per-lane EnqueueOrdinal inversions in the WorkItem readback order without starting workers or claiming pressure acceptance.", }, { Difference: "Normal enqueue and workflow-action materialization seed helper", LocalGate: "normal_enqueue_seed_scenario", Status: "covered", Notes: "Uses the connector event service to materialize batch-collection WorkItems, then readbacks counts without starting workers.", }, { Difference: "Progress-bucket readback for existing local pressure databases", LocalGate: "readback_progress_buckets", Status: "covered", Notes: "Samples state pairs, eligible lane heads, and current lease width over time from a local database without claiming test1 acceptance.", }, scenarioCatalogCoverage(), } report.RemainingCoverageGap = scenarioCatalogGaps() report.Results = []gateResult{result} report.Summary = summarizeReport(report.Results) if scenario, ok := findScenario(result.Gate.Scenario); ok { report.ScenarioCatalog = []scenarioDefinition{scenario} report.ScenarioSummary = summarizeCatalog(report.ScenarioCatalog) } if opts.SummaryOnly { report = compactSummaryReport(report) } if err := writeReport(opts.Out, report); err != nil { log.Fatal(err) } if result.Error != "" { os.Exit(1) } } func readback(ctx context.Context, db *sql.DB, opts runOptions) gateResult { result := gateResult{ Gate: gateConfig{ Name: "readback", Database: opts.ReadbackDB, Scenario: opts.Scenario, Explain: opts.Explain, ReadbackDB: true, }, } conn, err := db.Conn(ctx) if err != nil { result.Error = err.Error() return result } defer conn.Close() if err := useDatabase(ctx, conn, opts.ReadbackDB); err != nil { result.Error = err.Error() return result } result.Accepted, err = countEvents(ctx, conn, opts.PackID) if err != nil { result.Error = err.Error() return result } result.Materialized, err = countWorkItems(ctx, conn, opts.PackID, opts.Generation) if err != nil { result.Error = err.Error() return result } result.Processed, result.Skipped, result.Failed, result.Retry, result.Dead, result.Nonterminal, err = countStates(ctx, conn, opts.PackID, opts.Generation) if err != nil { result.Error = err.Error() return result } laneOrdinalAvailable, err := workItemLaneOrdinalAvailable(ctx, conn) if err != nil { result.Error = err.Error() return result } laneHeadReadbackAvailable := laneOrdinalAvailable if laneHeadReadbackAvailable { blankLaneFallbackNeeded, err := blankSourceLaneFallbackNeeded(ctx, conn, opts.PackID, opts.Generation) if err != nil { result.Error = err.Error() return result } laneHeadReadbackAvailable = !blankLaneFallbackNeeded } result.ClaimEligibleHeads, err = claimEligibleHeads(ctx, conn, opts.PackID, opts.Generation, laneHeadReadbackAvailable) if err != nil { result.Error = err.Error() return result } result.CurrentLeaseOwners, result.CurrentLeaseTokens, result.CurrentLeasedWork, result.CurrentLeasedLanes, err = currentLeaseCounts(ctx, conn, opts.PackID, opts.Generation) if err != nil { result.Error = err.Error() return result } result.MaxLeasedWork = result.CurrentLeasedWork result.MaxLeasedLanes = result.CurrentLeasedLanes result.MaxLeasedTokens = result.CurrentLeaseTokens result.StatePairs, err = statePairs(ctx, conn, opts.PackID, opts.Generation) if err != nil { result.Error = err.Error() return result } result.LaneOrderingOK, result.LaneOrderViolations, result.LaneOrderSamples, err = laneOrderSamples(ctx, conn, opts.PackID, opts.Generation, laneOrdinalAvailable) if err != nil { result.Error = err.Error() return result } if opts.Explain { result.ExplainAccessSummary, result.ExplainHasBroadScan, err = explainClaimSelect(ctx, conn, opts, laneHeadReadbackAvailable) if err != nil { result.Error = err.Error() } } if result.Error == "" && opts.BucketSeconds > 0 { buckets, err := observeProgressBuckets(ctx, conn, opts) if err != nil { result.Error = err.Error() return result } mergeProgressBuckets(&result, buckets) } return result } func observeProgressBuckets(ctx context.Context, conn *sql.Conn, opts runOptions) ([]progressBucket, error) { if opts.BucketSeconds <= 0 || opts.ObserveSeconds <= 0 { return nil, nil } laneOrdinalAvailable, err := workItemLaneOrdinalAvailable(ctx, conn) if err != nil { return nil, err } laneHeadReadbackAvailable := laneOrdinalAvailable if laneHeadReadbackAvailable { blankLaneFallbackNeeded, err := blankSourceLaneFallbackNeeded(ctx, conn, opts.PackID, opts.Generation) if err != nil { return nil, err } laneHeadReadbackAvailable = !blankLaneFallbackNeeded } started := time.Now() observeFor := time.Duration(opts.ObserveSeconds) * time.Second interval := time.Duration(opts.BucketSeconds) * time.Second var buckets []progressBucket for { bucket, err := readProgressBucket(ctx, conn, opts, laneHeadReadbackAvailable, time.Since(started).Seconds()) if err != nil { return nil, err } buckets = append(buckets, bucket) elapsed := time.Since(started) if elapsed >= observeFor { break } sleepFor := interval if remaining := observeFor - elapsed; remaining < sleepFor { sleepFor = remaining } timer := time.NewTimer(sleepFor) select { case <-ctx.Done(): timer.Stop() return nil, ctx.Err() case <-timer.C: } } return buckets, nil } func readProgressBucket(ctx context.Context, conn *sql.Conn, opts runOptions, laneHeadReadbackAvailable bool, elapsedSeconds float64) (progressBucket, error) { var bucket progressBucket bucket.ElapsedSeconds = elapsedSeconds var err error bucket.Processed, bucket.Skipped, bucket.Failed, bucket.Retry, bucket.Dead, bucket.Nonterminal, err = countStates(ctx, conn, opts.PackID, opts.Generation) if err != nil { return bucket, err } bucket.ClaimEligibleHeads, err = claimEligibleHeads(ctx, conn, opts.PackID, opts.Generation, laneHeadReadbackAvailable) if err != nil { return bucket, err } bucket.CurrentLeaseOwners, bucket.CurrentLeaseTokens, bucket.CurrentLeasedWork, bucket.CurrentLeasedLanes, err = currentLeaseCounts(ctx, conn, opts.PackID, opts.Generation) if err != nil { return bucket, err } bucket.StatePairs, err = statePairs(ctx, conn, opts.PackID, opts.Generation) if err != nil { return bucket, err } return bucket, nil } func mergeProgressBuckets(result *gateResult, buckets []progressBucket) { result.Buckets = buckets for _, bucket := range buckets { if bucket.CurrentLeasedWork > result.MaxLeasedWork { result.MaxLeasedWork = bucket.CurrentLeasedWork } if bucket.CurrentLeasedLanes > result.MaxLeasedLanes { result.MaxLeasedLanes = bucket.CurrentLeasedLanes } if bucket.CurrentLeaseTokens > result.MaxLeasedTokens { result.MaxLeasedTokens = bucket.CurrentLeaseTokens } } } func summarizeReport(results []gateResult) reportSummary { summary := reportSummary{ ResultCount: len(results), Status: "catalog_only", HarnessOnly: true, AcceptanceSurface: "local_only_no_test1", Notes: []string{ "Local harness summary only; no test1, staging, prod, or provider acceptance is claimed.", }, } if len(results) == 0 { return summary } summary.Status = "ok" for _, result := range results { gate := summarizeGate(result) summary.Gates = append(summary.Gates, gate) summary.TotalAccepted += result.Accepted summary.TotalMaterialized += result.Materialized summary.TotalProcessed += result.Processed summary.TotalSkipped += result.Skipped summary.TotalFailed += result.Failed summary.TotalRetry += result.Retry summary.TotalDead += result.Dead summary.TotalNonterminal += result.Nonterminal if result.ClaimEligibleHeads > summary.MaxClaimEligibleHeads { summary.MaxClaimEligibleHeads = result.ClaimEligibleHeads } if result.MaxLeasedWork > summary.MaxLeasedWork { summary.MaxLeasedWork = result.MaxLeasedWork } if result.MaxLeasedLanes > summary.MaxLeasedLanes { summary.MaxLeasedLanes = result.MaxLeasedLanes } if result.MaxLeasedTokens > summary.MaxLeasedTokens { summary.MaxLeasedTokens = result.MaxLeasedTokens } if len(result.Buckets) > summary.MaxBucketCount { summary.MaxBucketCount = len(result.Buckets) } switch { case result.Error != "": summary.ErrorCount++ case result.Nonterminal > 0: summary.NonterminalCount++ default: summary.CompletedCount++ } if result.ExplainHasBroadScan { summary.BroadScanCount++ } if !result.LaneOrderingOK { summary.LaneOrderFailureCount++ } } switch { case summary.ErrorCount > 0: summary.Status = "error" case summary.BroadScanCount > 0: summary.Status = "broad_scan" case summary.LaneOrderFailureCount > 0: summary.Status = "lane_order_violation" case summary.NonterminalCount > 0: summary.Status = "nonterminal" } return summary } func summarizeGate(result gateResult) gateSummary { gate := gateSummary{ Name: result.Gate.Name, Database: result.Gate.Database, Scenario: result.Gate.Scenario, Status: "ok", Accepted: result.Accepted, Materialized: result.Materialized, Terminal: result.Processed + result.Skipped + result.Failed + result.Dead, Nonterminal: result.Nonterminal, ClaimEligibleHeads: result.ClaimEligibleHeads, MaxLeasedWork: result.MaxLeasedWork, MaxLeasedLanes: result.MaxLeasedLanes, MaxLeasedTokens: result.MaxLeasedTokens, BucketCount: len(result.Buckets), LaneOrderingOK: result.LaneOrderingOK, ExplainHasBroadScan: result.ExplainHasBroadScan, Error: result.Error, } if gate.Error != "" { gate.Status = "error" return gate } switch { case result.ExplainHasBroadScan: gate.Status = "broad_scan" case !result.LaneOrderingOK: gate.Status = "lane_order_violation" case result.Nonterminal > 0: gate.Status = "nonterminal" } if len(result.Buckets) > 0 { last := result.Buckets[len(result.Buckets)-1] gate.LastBucketProcessed = last.Processed gate.LastBucketSkipped = last.Skipped gate.LastBucketFailed = last.Failed gate.LastBucketRetry = last.Retry gate.LastBucketDead = last.Dead gate.LastBucketPending = last.Nonterminal } return gate } func summarizeCatalog(scenarios []scenarioDefinition) catalogSummary { summary := catalogSummary{ScenarioCount: len(scenarios)} seedModes := map[string]int{} flags := map[string]int{} for _, scenario := range scenarios { if isSeedableScenario(scenario) { summary.SeedableCount++ } if scenario.RuntimeRequired { summary.RuntimeRequiredCount++ } mode := strings.TrimSpace(scenario.SeedMode) if mode == "" { mode = "unspecified" } seedModes[mode]++ for _, flag := range scenario.Flags { flags[flag]++ } } summary.SeedModes = sortedNamedCounts(seedModes) summary.Flags = sortedNamedCounts(flags) return summary } func sortedNamedCounts(values map[string]int) []namedCount { if len(values) == 0 { return nil } names := make([]string, 0, len(values)) for name := range values { names = append(names, name) } sort.Strings(names) out := make([]namedCount, 0, len(names)) for _, name := range names { out = append(out, namedCount{Name: name, Count: values[name]}) } return out } func compactSummaryReport(report runReport) runReport { report.Results = nil report.ScenarioCatalog = nil return report } func seedScenario(ctx context.Context, db *sql.DB, opts *runOptions) gateResult { scenario, ok := findScenario(opts.SeedScenario) if !ok { return gateResult{Error: fmt.Sprintf("unknown -seed-scenario %q; known scenarios: %s", opts.SeedScenario, strings.Join(scenarioNames(), ", "))} } if !isSeedableScenario(scenario) { return gateResult{Error: fmt.Sprintf("scenario %q is not seed-only; worker execution remains successor work", scenario.Name)} } dbName := strings.TrimSpace(opts.SeedDB) if dbName == "" { dbName = generatedSeedDBName(scenario.Name, time.Now().UTC()) } result := gateResult{ Gate: gateConfig{ Name: "seed_scenario", Database: dbName, Scenario: scenario.Name, Explain: opts.Explain, SeedDB: true, }, } if err := createSeedDatabase(ctx, db, dbName); err != nil { result.Error = err.Error() return result } conn, err := db.Conn(ctx) if err != nil { result.Error = err.Error() return result } defer conn.Close() if err := useDatabase(ctx, conn, dbName); err != nil { result.Error = err.Error() return result } if err := createSchema(ctx, conn, scenario); err != nil { result.Error = err.Error() return result } dirtyRows := 0 if scenario.SeedMode == "normal_enqueue" { seedDB, err := openSeedDatabase(ctx, opts.DSN, dbName) if err != nil { result.Error = err.Error() return result } defer seedDB.Close() if err := seedNormalEnqueue(ctx, seedDB, scenario, time.Now().UTC().Add(-time.Minute)); err != nil { result.Error = err.Error() return result } opts.Generation = scenarioSeedGeneration(scenario) } else { var err error dirtyRows, err = seedDirectWorkItems(ctx, conn, scenario, *opts) if err != nil { result.Error = err.Error() return result } } opts.ReadbackDB = dbName opts.Scenario = scenario.Name result = readback(ctx, db, *opts) result.Gate.Name = "seed_scenario" result.Gate.SeedDB = true result.DirtyRowsSeeded = dirtyRows return result } func createSeedDatabase(ctx context.Context, db *sql.DB, name string) error { if !safeIdentifier.MatchString(name) { return fmt.Errorf("unsafe database name %q", name) } if len(name) > maxIdentifierLength { return fmt.Errorf("database name %q exceeds %d characters", name, maxIdentifierLength) } _, err := db.ExecContext(ctx, "CREATE DATABASE `"+name+"`") return err } func isSeedableScenario(scenario scenarioDefinition) bool { return scenario.SeedMode == "direct_work_items" || scenario.SeedMode == "normal_enqueue" } func scenarioSeedGeneration(scenario scenarioDefinition) int64 { if scenario.SeedMode == "normal_enqueue" { return normalEnqueueGeneration } return defaultGeneration } func openSeedDatabase(ctx context.Context, rootDSN, dbName string) (*sql.DB, error) { dsn, err := dsnWithDatabase(rootDSN, dbName) if err != nil { return nil, err } db, err := sql.Open("mysql", dsn) if err != nil { return nil, err } if err := db.PingContext(ctx); err != nil { _ = db.Close() return nil, err } return db, nil } func dsnWithDatabase(dsn, dbName string) (string, error) { if !safeIdentifier.MatchString(dbName) { return "", fmt.Errorf("unsafe database name %q", dbName) } prefix, suffix, _ := strings.Cut(dsn, "?") if suffix != "" { suffix = "?" + suffix } slash := strings.LastIndex(prefix, "/") if slash < 0 { return "", fmt.Errorf("dsn has no database separator") } return prefix[:slash+1] + dbName + suffix, nil } func createSchema(ctx context.Context, conn *sql.Conn, scenario scenarioDefinition) error { if scenario.SeedMode == "normal_enqueue" { return connectorfixture.CreateSystemTablesContext(ctx, conn, "_System_InstalledPack", "_System_ConnectorInboundRoute", "_System_ConnectorIngressControl", "_System_ConnectorEvent", "_System_ConnectorWorkItem", "_System_ConnectorProjectionProtocol") } // Direct seeds intentionally model legacy/malformed schemas; the actual // enqueue walking slice above always uses the canonical current catalog. laneOrdinalColumn := "LaneOrdinal BIGINT NOT NULL" if hasFlag(scenario, "default_lane_ordinal_column") { laneOrdinalColumn = "LaneOrdinal BIGINT NOT NULL DEFAULT 0" } statements := []string{` CREATE TABLE _System_InstalledPack ( Id VARCHAR(255) NOT NULL PRIMARY KEY, PackInstallId VARCHAR(255) NOT NULL, Name VARCHAR(255) NOT NULL, Category VARCHAR(64) NOT NULL, Status VARCHAR(64) NOT NULL, AppliedPayloadHash VARCHAR(255) NULL, DesiredPayloadHash VARCHAR(255) NULL, PackVersion VARCHAR(255) NULL, IsDeleted TINYINT NOT NULL DEFAULT 0, CreatedDate DATETIME NOT NULL, LastModifiedDate DATETIME NOT NULL, UNIQUE KEY uniq_installed_pack_install_id (PackInstallId) );`, ` CREATE TABLE _System_ConnectorInboundRoute ( Id VARCHAR(255) NOT NULL PRIMARY KEY, InstalledPackId VARCHAR(255) NOT NULL, EventKind VARCHAR(255) NOT NULL, Version BIGINT NOT NULL, FlowDeveloperName VARCHAR(255) NOT NULL DEFAULT 'NoopFlow__c', InvocationMode VARCHAR(64) NOT NULL DEFAULT 'per_event', BatchInputName VARCHAR(128) NOT NULL DEFAULT '', BatchResultOutputName VARCHAR(128) NOT NULL DEFAULT '', MaxBatchSize INT NOT NULL DEFAULT 0, QueueClass VARCHAR(32) NOT NULL, FlowExecutionFilter JSON NULL, IsDeleted TINYINT NOT NULL DEFAULT 0, IsCurrent TINYINT NOT NULL DEFAULT 1, CreatedDate DATETIME NOT NULL, LastModifiedDate DATETIME NOT NULL, KEY idx_connector_inbound_route_current (InstalledPackId, EventKind, Version, IsDeleted, IsCurrent) );`, ` CREATE TABLE _System_ConnectorIngressControl ( Id VARCHAR(255) NOT NULL PRIMARY KEY, Namespace VARCHAR(255) NOT NULL, PackInstallId VARCHAR(255) NOT NULL DEFAULT '', Channel VARCHAR(255) NOT NULL, Mode VARCHAR(32) NOT NULL DEFAULT 'active', FenceEpoch BIGINT NOT NULL DEFAULT 0, RetryAfterSeconds INT NOT NULL DEFAULT 120, Reason VARCHAR(255) NULL, ExpiresAt DATETIME NULL, IsDeleted TINYINT NOT NULL DEFAULT 0, CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uniq_connector_ingress_control_scope (Namespace, PackInstallId, Channel) );`, ` CREATE TABLE _System_ConnectorEvent ( Id VARCHAR(255) NOT NULL PRIMARY KEY, Namespace VARCHAR(255) NOT NULL DEFAULT 'local-pressure', InstalledPackId VARCHAR(255) NOT NULL, EventKind VARCHAR(255) NOT NULL DEFAULT 'noop.created', IdempotencyKey VARCHAR(255) NOT NULL DEFAULT '', Status VARCHAR(32) NOT NULL, ExecutionOwner VARCHAR(32) NOT NULL DEFAULT 'event', QueueClass VARCHAR(32) NOT NULL DEFAULT 'default', LastModifiedDate DATETIME NOT NULL, NextAttemptAt DATETIME NULL, RetryCount INT NOT NULL DEFAULT 0, LastError TEXT NULL, LastErrorAt DATETIME NULL, ClaimedBy VARCHAR(255) NULL, ClaimedAt DATETIME NULL, ClaimToken VARCHAR(255) NULL, ProcessedAt DATETIME NULL, ReplayedAt DATETIME NULL, ReplayOfId VARCHAR(255) NULL, RouteVersion BIGINT NULL DEFAULT 1, CompanionPackInstallId VARCHAR(255) NULL, CompanionFlowVersionId VARCHAR(255) NULL, SourceObjectKind VARCHAR(64) NULL, SourceObjectID VARCHAR(255) NULL, SourceLaneKey VARCHAR(64) NULL, Payload JSON NULL, EnqueueOrdinal BIGINT NOT NULL, CreatedDate DATETIME NOT NULL, DeliveryId VARCHAR(255) NULL, DeliverySequence BIGINT NOT NULL, KEY idx_event_pack_status (InstalledPackId, Status, LastModifiedDate), KEY idx_event_source_lane (InstalledPackId, SourceLaneKey, EnqueueOrdinal) );`, ` CREATE TABLE _System_ConnectorWorkItem ( WorkId VARCHAR(255) NOT NULL PRIMARY KEY, EventId VARCHAR(255) NOT NULL, InstalledPackId VARCHAR(255) NOT NULL, WorkType VARCHAR(64) NOT NULL, QueueClass VARCHAR(32) NOT NULL, State VARCHAR(32) NOT NULL, RunAfter DATETIME NOT NULL, Attempt INT NOT NULL, OwnershipEpoch BIGINT NOT NULL DEFAULT 0, FailureCount INT NOT NULL DEFAULT 0, FailureClass VARCHAR(64) NULL, FailureSummary TEXT NULL, LastFailureAt DATETIME NULL, ReplayedAt DATETIME NULL, ReplayOfWorkId VARCHAR(255) NULL, ` + laneOrdinalColumn + `, LeaseOwner VARCHAR(255) NULL, LeaseToken VARCHAR(255) NULL, LeaseUntil DATETIME NULL, SourceLaneKey VARCHAR(64) NULL, TargetMutationLaneKey VARCHAR(64) NULL, Generation BIGINT NOT NULL, ProjectionKey VARCHAR(255) NULL, ProjectionActiveKey VARCHAR(255) NULL, TerminalAt DATETIME NULL, CreatedDate DATETIME NOT NULL, LastModifiedDate DATETIME NOT NULL, UNIQUE KEY uniq_connector_work_item_event_type (EventId, WorkType), UNIQUE KEY uniq_connector_work_item_replay_of (ReplayOfWorkId), UNIQUE KEY uniq_connector_work_item_projection_active (ProjectionActiveKey), KEY idx_connector_work_item_queue_claim_gen (InstalledPackId, Generation, QueueClass, State, RunAfter, WorkId), KEY idx_connector_work_item_lease_due_gen (InstalledPackId, Generation, State, LeaseUntil, WorkId), KEY idx_connector_work_item_source_lane_gen (InstalledPackId, Generation, SourceLaneKey, State, LaneOrdinal, RunAfter, WorkId), KEY idx_connector_work_item_target_lane (InstalledPackId, TargetMutationLaneKey, State, RunAfter, WorkId), KEY idx_connector_work_item_projection_history (ProjectionKey, TerminalAt, WorkId) );`} for _, stmt := range statements { if _, err := conn.ExecContext(ctx, stmt); err != nil { return err } } return nil } func seedDirectWorkItems(ctx context.Context, conn *sql.Conn, scenario scenarioDefinition, opts runOptions) (int, error) { now := time.Now().UTC().Add(-time.Minute) tx, err := conn.BeginTx(ctx, nil) if err != nil { return 0, err } defer tx.Rollback() dirtyRows := 0 if hasFlag(scenario, "dirty_preexisting_rows") { dirtyRows, err = seedDirtyPreexistingRows(ctx, tx, now, opts) if err != nil { return 0, err } } eventStmt, err := tx.PrepareContext(ctx, `INSERT INTO _System_ConnectorEvent (Id, InstalledPackId, Status, LastModifiedDate, NextAttemptAt, SourceLaneKey, Payload, EnqueueOrdinal, CreatedDate, DeliveryId, DeliverySequence) VALUES (?, ?, 'pending', ?, NULL, ?, JSON_OBJECT('objectKind','contact','sourceID',?,'payload',JSON_OBJECT('id', ?)), ?, ?, NULL, ?)`) if err != nil { return 0, err } defer eventStmt.Close() workInsertSQL := `INSERT INTO _System_ConnectorWorkItem (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, Attempt, LaneOrdinal, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey, TargetMutationLaneKey, Generation, CreatedDate, LastModifiedDate) VALUES (?, ?, ?, 'connector_event', 'default', ?, ?, 0, ?, NULL, NULL, NULL, ?, '', ?, ?, ?)` if hasFlag(scenario, "default_lane_ordinal_column") { workInsertSQL = `INSERT INTO _System_ConnectorWorkItem (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, Attempt, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey, TargetMutationLaneKey, Generation, CreatedDate, LastModifiedDate) VALUES (?, ?, ?, 'connector_event', 'default', ?, ?, 0, NULL, NULL, NULL, ?, '', ?, ?, ?)` } workStmt, err := tx.PrepareContext(ctx, workInsertSQL) if err != nil { return 0, err } defer workStmt.Close() laneSeq := map[string]int{} for i := 0; i < scenario.Events; i++ { lane := fmt.Sprintf("lane-%04d", i%scenario.Lanes) laneSeq[lane]++ eventID := fmt.Sprintf("event-%06d", i) workID := fmt.Sprintf("work-%06d", i) if hasFlag(scenario, "scrambled_work_id") { workID = fmt.Sprintf("work-%06d", scenario.Events-i) } created := now.Add(time.Duration(i) * time.Microsecond) if hasFlag(scenario, "same_created_date") { created = now } state := "pending" laneOrdinal := laneSeq[lane] if hasFlag(scenario, "zero_lane_ordinal") { laneOrdinal = 0 } if _, err := eventStmt.ExecContext(ctx, eventID, opts.PackID, now, lane, lane, lane, laneSeq[lane], created, i); err != nil { return 0, err } if hasFlag(scenario, "default_lane_ordinal_column") { if _, err := workStmt.ExecContext(ctx, workID, eventID, opts.PackID, state, now, lane, opts.Generation, created, now); err != nil { return 0, err } continue } if _, err := workStmt.ExecContext(ctx, workID, eventID, opts.PackID, state, now, laneOrdinal, lane, opts.Generation, created, now); err != nil { return 0, err } } if err := tx.Commit(); err != nil { return 0, err } return dirtyRows, nil } func seedNormalEnqueue(ctx context.Context, db *sql.DB, scenario scenarioDefinition, now time.Time) error { eventKind := "noop.created" if hasFlag(scenario, "workflow_action") { eventKind = "workflow.action.execute" } if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,1,0)`, defaultInstalledPackID); err != nil { return err } if _, err := db.ExecContext(ctx, ` INSERT INTO _System_InstalledPack (Id, PackInstallId, Name, Category, Status, AppliedPayloadHash, DesiredPayloadHash, PackVersion, IsDeleted, CreatedDate, LastModifiedDate) VALUES ('pack-row-local-pressure', ?, 'local-pressure', ?, 'active', 'hash-local-pressure', 'hash-local-pressure', 'v-local', 0, ?, ?)`, defaultInstalledPackID, string(packstate.ConnectorPack), now, now); err != nil { return err } if _, err := db.ExecContext(ctx, ` INSERT INTO _System_ConnectorInboundRoute (Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName, InvocationMode, BatchInputName, BatchResultOutputName, MaxBatchSize, QueueClass, FlowExecutionFilter, IsDeleted, IsCurrent, CreatedDate, LastModifiedDate) VALUES ('route-local-pressure', 'local-pressure', ?, ?, 1, 'NoopFlow__c', ?, ?, ?, ?, ?, NULL, 0, 1, ?, ?)`, defaultInstalledPackID, eventKind, string(connector_inbound_routing.InvocationModeBatchCollection), connector_inbound_routing.DefaultBatchInputName, connector_inbound_routing.DefaultBatchResultOutputName, connector_inbound_routing.DefaultRouteMaxBatchSize, connector_inbound_routing.QueueClassBulk, now, now); err != nil { return err } svc := connector_events.NewService(db, pressureIDGen(), func() time.Time { return now }) for start := 0; start < scenario.Events; start += connector_inbound_routing.DefaultRouteMaxBatchSize { end := start + connector_inbound_routing.DefaultRouteMaxBatchSize if end > scenario.Events { end = scenario.Events } events := make([]connector_events.BatchEvent, 0, end-start) for i := start; i < end; i++ { sourceID := fmt.Sprintf("source-%04d", i%scenario.Lanes) payload, err := connector_inbound.EncodeQueueEnvelope("contact", sourceID, json.RawMessage(fmt.Sprintf(`{"id":%q}`, sourceID))) if err != nil { return err } events = append(events, connector_events.BatchEvent{ EventKind: eventKind, IdempotencyKey: fmt.Sprintf("normal-ikey-%06d", i), Payload: payload, RouteVersion: 1, SourceLaneObjectKind: "contact", SourceLaneID: sourceID, DeliverySequence: i, }) } if _, _, err := svc.EnqueueBatch(ctx, "local-pressure", defaultInstalledPackID, fmt.Sprintf("normal-delivery-%03d", start/connector_inbound_routing.DefaultRouteMaxBatchSize), events); err != nil { return err } } return nil } func pressureIDGen() func() string { var next atomic.Int64 return func() string { return fmt.Sprintf("event-%06d", next.Add(1)-1) } } func seedDirtyPreexistingRows(ctx context.Context, tx *sql.Tx, now time.Time, opts runOptions) (int, error) { eventStmt, err := tx.PrepareContext(ctx, `INSERT INTO _System_ConnectorEvent (Id, InstalledPackId, Status, LastModifiedDate, NextAttemptAt, SourceLaneKey, Payload, EnqueueOrdinal, CreatedDate, DeliveryId, DeliverySequence) VALUES (?, ?, ?, ?, NULL, ?, JSON_OBJECT('objectKind','contact','sourceID',?), ?, ?, NULL, ?)`) if err != nil { return 0, err } defer eventStmt.Close() workStmt, err := tx.PrepareContext(ctx, `INSERT INTO _System_ConnectorWorkItem (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, Attempt, LaneOrdinal, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey, TargetMutationLaneKey, Generation, CreatedDate, LastModifiedDate) VALUES (?, ?, ?, 'connector_event', 'default', ?, ?, 0, ?, NULL, NULL, NULL, ?, '', ?, ?, ?)`) if err != nil { return 0, err } defer workStmt.Close() type dirtyGroup struct { packID string generation int64 statuses []string states []string } groups := []dirtyGroup{ {packID: opts.PackID + "-old-generation", generation: opts.Generation - 1, statuses: []string{"pending", "processed", "dead"}, states: []string{"pending", "processed", "dead", "blocked"}}, {packID: "pack-dirty-preexisting", generation: opts.Generation, statuses: []string{"pending", "processed", "dead"}, states: []string{"pending", "processed", "dead", "blocked"}}, } inserted := 0 for groupIdx, group := range groups { for i := 0; i < 48; i++ { lane := fmt.Sprintf("dirty-lane-%02d", i%8) eventID := fmt.Sprintf("dirty-event-%02d-%03d", groupIdx, i) workID := fmt.Sprintf("dirty-work-%02d-%03d", groupIdx, i) status := group.statuses[i%len(group.statuses)] state := group.states[i%len(group.states)] created := now.Add(-time.Duration(i+1) * time.Second) if _, err := eventStmt.ExecContext(ctx, eventID, group.packID, status, created, lane, lane, i+1, created, i+1); err != nil { return inserted, err } if _, err := workStmt.ExecContext(ctx, workID, eventID, group.packID, state, created, i+1, lane, group.generation, created, created); err != nil { return inserted, err } inserted++ } } return inserted, nil } func baseReport(opts runOptions) runReport { return runReport{ Repo: runGitOrEmpty("rev-parse", "--show-toplevel"), Branch: runGitOrEmpty("branch", "--show-current"), Head: runGitOrEmpty("rev-parse", "HEAD"), DSN: redactDSN(opts.DSN), GeneratedAt: time.Now().UTC().Format(time.RFC3339), } } func scenarioCatalogCoverage() differenceCoverage { return differenceCoverage{ Difference: "PR #43 scenario definitions and presets", LocalGate: "scenario_catalog", Status: "covered", Notes: "Metadata-only catalog extracted from PR #43 gate definitions; it does not seed rows, start workers, run pressure, or classify PASS/BLOCK.", } } func scenarioCatalogGaps() []string { return []string{ "Seed helper covers direct WorkItem and normal enqueue/workflow-action materialization scenarios only.", "Worker/CentralClaimScheduler execution is intentionally out of this slice.", "Claim latency, progress buckets, and worker-claim lane-order streams are intentionally out of this slice.", "No test1/prod/staging/provider acceptance is claimed by this local tooling slice.", } } func scenarioCatalog() []scenarioDefinition { return []scenarioDefinition{ { Name: "clean_800_events_120_lanes", Source: "PR43 default gate worker16_800_events_120_lanes", Events: 800, Lanes: 120, Workers: 16, SeedMode: "direct_work_items", Purpose: "Baseline clean materialized WorkItem shape for local readback and future seed helpers.", }, { Name: "zero_laneordinal_800_events_120_lanes", Source: "PR43 default gate zero_laneordinal_800_events_120_lanes", Events: 800, Lanes: 120, Workers: 16, SeedMode: "direct_work_items", Flags: []string{"zero_lane_ordinal", "same_created_date", "scrambled_work_id"}, Purpose: "Models the post-hotpatch zero LaneOrdinal ordering shape without claiming runtime drain.", }, { Name: "default_laneordinal_800_events_120_lanes", Source: "PR43 default gate default_laneordinal_800_events_120_lanes", Events: 800, Lanes: 120, Workers: 16, SeedMode: "direct_work_items", Flags: []string{"default_lane_ordinal_column", "same_created_date", "scrambled_work_id"}, Purpose: "Captures the supported LaneOrdinal defaulting fixture shape for a later schema/seed slice.", }, { Name: "dirty_preexisting_rows_800_events_120_lanes", Source: "PR43 default gate dirty_preexisting_rows_800_events_120_lanes", Events: 800, Lanes: 120, Workers: 16, SeedMode: "direct_work_items", Flags: []string{"zero_lane_ordinal", "same_created_date", "scrambled_work_id", "dirty_preexisting_rows"}, Purpose: "Preserves the out-of-scope dirty-row fixture definition while keeping this PR read/metadata-only.", }, { Name: "large_pending_1000_events_501_lanes", Source: "PR43 default gate large_pending_1000_events_501_lanes", Events: 1000, Lanes: 501, Workers: 16, SeedMode: "direct_work_items", Flags: []string{"explain"}, Purpose: "Records the large pending backlog scenario for future access-path and seed validation.", }, { Name: "normal_enqueue_worker_path_800_events_120_lanes", Source: "PR43 default gate normal_enqueue_worker_path_800_events_120_lanes", Events: 800, Lanes: 120, Workers: 16, SeedMode: "normal_enqueue", RuntimeRequired: true, Flags: []string{"worker_path", "normal_enqueue", "explain"}, Purpose: "Seeds the normal EnqueueBatch materialization shape for local readback; worker execution remains successor work.", }, { Name: "workflow_action_worker_path_800_events_120_lanes", Source: "PR43 default gate normal_workflow_action_worker_path_800_events_120_lanes", Events: 800, Lanes: 120, Workers: 16, SeedMode: "normal_enqueue", RuntimeRequired: true, Flags: []string{"worker_path", "normal_enqueue", "workflow_action", "explain"}, Purpose: "Seeds the workflow-action EnqueueBatch materialization shape for local readback; worker execution remains successor work.", }, { Name: "b2b97653_tail_after_partial_skips", Source: "PR43 flag b2b-local-repro", Events: 800, Lanes: 120, Workers: 16, SeedMode: "direct_work_items", RuntimeRequired: true, Flags: []string{"zero_lane_ordinal", "same_created_date", "scrambled_work_id", "default_lane_ordinal_column", "skip_terminal", "forced_max_processing_rows=3", "stop_after_skipped=54", "observe_after_stop_seconds=30", "bucket_seconds=10", "reverse_event_ordinal", "explain"}, Purpose: "Preserves the deterministic local tail-shape repro preset as catalog metadata, not a PASS/BLOCK assertion.", }, } } func findScenario(name string) (scenarioDefinition, bool) { needle := strings.TrimSpace(name) if needle == "" { return scenarioDefinition{}, false } for _, scenario := range scenarioCatalog() { if scenario.Name == needle { return scenario, true } } return scenarioDefinition{}, false } func scenarioNames() []string { scenarios := scenarioCatalog() names := make([]string, 0, len(scenarios)) for _, scenario := range scenarios { names = append(names, scenario.Name) } return names } func hasFlag(scenario scenarioDefinition, flag string) bool { for _, candidate := range scenario.Flags { if candidate == flag { return true } } return false } func generatedSeedDBName(scenarioName string, now time.Time) string { timestamp := now.Format(seedDBTimestampLayout) fragmentBudget := maxIdentifierLength - len(seedDBPrefix) - 1 - len(timestamp) if fragmentBudget < 1 { fragmentBudget = 1 } return fmt.Sprintf("%s%s_%s", seedDBPrefix, compactIdentifier(scenarioName, fragmentBudget), timestamp) } func compactIdentifier(value string, maxLen int) string { var b strings.Builder for _, r := range value { switch { case r >= 'a' && r <= 'z': b.WriteRune(r) case r >= 'A' && r <= 'Z': b.WriteRune(r) case r >= '0' && r <= '9': b.WriteRune(r) default: b.WriteByte('_') } if b.Len() >= maxLen { break } } out := strings.Trim(b.String(), "_") if out == "" { return "scenario" } return out } func writeReport(out string, report runReport) error { data, err := json.MarshalIndent(report, "", " ") if err != nil { return err } if out == "" { fmt.Println(string(data)) return nil } return os.WriteFile(out, append(data, '\n'), 0o644) } func useDatabase(ctx context.Context, conn *sql.Conn, name string) error { if !safeIdentifier.MatchString(name) { return fmt.Errorf("unsafe database name %q", name) } _, err := conn.ExecContext(ctx, "USE `"+name+"`") return err } func countEvents(ctx context.Context, conn *sql.Conn, packID string) (int, error) { return countScalar(ctx, conn, `SELECT COUNT(*) FROM _System_ConnectorEvent WHERE InstalledPackId=?`, packID) } func countWorkItems(ctx context.Context, conn *sql.Conn, packID string, generation int64) (int, error) { return countScalar(ctx, conn, `SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId=? AND Generation=?`, packID, generation) } func countStates(ctx context.Context, conn *sql.Conn, packID string, generation int64) (processed, skipped, failed, retry, dead, nonterminal int, err error) { rows, err := conn.QueryContext(ctx, `SELECT State, COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId=? AND Generation=? GROUP BY State`, packID, generation) if err != nil { return 0, 0, 0, 0, 0, 0, err } defer rows.Close() for rows.Next() { var state string var count int if err := rows.Scan(&state, &count); err != nil { return 0, 0, 0, 0, 0, 0, err } switch state { case "processed": processed = count case "skipped": skipped = count case "failed": failed = count case "retry": retry = count nonterminal += count case "dead": dead = count default: nonterminal += count } } if err := rows.Err(); err != nil { return 0, 0, 0, 0, 0, 0, err } return processed, skipped, failed, retry, dead, nonterminal, nil } func workItemLaneOrdinalAvailable(ctx context.Context, conn *sql.Conn) (bool, error) { count, err := countScalar(ctx, conn, ` SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_ConnectorWorkItem' AND COLUMN_NAME = 'LaneOrdinal'`) if err != nil { return false, err } return count > 0, nil } func blankSourceLaneFallbackNeeded(ctx context.Context, conn *sql.Conn, packID string, generation int64) (bool, error) { now := time.Now().UTC() count, err := countScalar(ctx, conn, ` SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ? AND Generation = ? AND State IN ('pending', 'retry') AND RunAfter <= ? AND COALESCE(SourceLaneKey, '') = ''`, packID, generation, now) if err != nil { return false, err } return count > 0, nil } func claimEligibleHeads(ctx context.Context, conn *sql.Conn, packID string, generation int64, laneHeadReadbackAvailable bool) (int, error) { now := time.Now().UTC() if !laneHeadReadbackAvailable { return countScalar(ctx, conn, ` SELECT COUNT(*) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ? AND Generation = ? AND State IN ('pending', 'retry') AND RunAfter <= ?`, packID, generation, now) } return countScalar(ctx, conn, ` SELECT COUNT(*) FROM _System_ConnectorWorkItem w WHERE w.InstalledPackId = ? AND w.Generation = ? AND w.State IN ('pending', 'retry') AND w.RunAfter <= ? AND NOT EXISTS ( SELECT 1 FROM _System_ConnectorWorkItem prior WHERE prior.InstalledPackId = w.InstalledPackId AND prior.Generation = w.Generation AND prior.State IN ('pending', 'retry') AND prior.RunAfter <= ? AND COALESCE(prior.SourceLaneKey, '') = COALESCE(w.SourceLaneKey, '') AND ( prior.LaneOrdinal < w.LaneOrdinal OR (prior.LaneOrdinal = w.LaneOrdinal AND prior.RunAfter < w.RunAfter) OR (prior.LaneOrdinal = w.LaneOrdinal AND prior.RunAfter = w.RunAfter AND prior.WorkId < w.WorkId) ) )`, packID, generation, now, now) } func currentLeaseCounts(ctx context.Context, conn *sql.Conn, packID string, generation int64) (owners, tokens, workItems, lanes int, err error) { err = conn.QueryRowContext(ctx, ` SELECT COUNT(DISTINCT COALESCE(LeaseOwner, '')), COUNT(DISTINCT COALESCE(LeaseToken, '')), COUNT(*), COUNT(DISTINCT COALESCE(SourceLaneKey, '')) FROM _System_ConnectorWorkItem WHERE InstalledPackId = ? AND Generation = ? AND State = 'processing' AND LeaseUntil > ?`, packID, generation, time.Now().UTC()).Scan(&owners, &tokens, &workItems, &lanes) return owners, tokens, workItems, lanes, err } func statePairs(ctx context.Context, conn *sql.Conn, packID string, generation int64) ([]statePairBucket, error) { rows, err := conn.QueryContext(ctx, ` SELECT w.State, COALESCE(e.Status, ''), COUNT(*) FROM _System_ConnectorWorkItem w LEFT JOIN _System_ConnectorEvent e ON e.Id = w.EventId WHERE w.InstalledPackId = ? AND w.Generation = ? GROUP BY w.State, COALESCE(e.Status, '') ORDER BY COUNT(*) DESC, w.State, COALESCE(e.Status, '')`, packID, generation) if err != nil { return nil, err } defer rows.Close() var pairs []statePairBucket for rows.Next() { var pair statePairBucket if err := rows.Scan(&pair.WorkState, &pair.EventStatus, &pair.Count); err != nil { return nil, err } pairs = append(pairs, pair) } return pairs, rows.Err() } type laneOrderRecord struct { Lane string LaneOrdinal int64 RunAfter time.Time WorkID string EventID string EnqueueOrdinal int64 } func laneOrderSamples(ctx context.Context, conn *sql.Conn, packID string, generation int64, laneOrdinalAvailable bool) (bool, int, []string, error) { if !laneOrdinalAvailable { return true, 0, nil, nil } rows, err := conn.QueryContext(ctx, ` SELECT COALESCE(w.SourceLaneKey, ''), w.LaneOrdinal, w.RunAfter, w.WorkId, w.EventId, COALESCE(e.EnqueueOrdinal, 0) FROM _System_ConnectorWorkItem w LEFT JOIN _System_ConnectorEvent e ON e.Id = w.EventId WHERE w.InstalledPackId = ? AND w.Generation = ? AND w.State IN ('pending', 'retry') AND w.RunAfter <= ? ORDER BY COALESCE(w.SourceLaneKey, ''), w.LaneOrdinal ASC, w.RunAfter ASC, w.WorkId ASC`, packID, generation, time.Now().UTC()) if err != nil { return false, 0, nil, err } defer rows.Close() var records []laneOrderRecord for rows.Next() { var record laneOrderRecord if err := rows.Scan(&record.Lane, &record.LaneOrdinal, &record.RunAfter, &record.WorkID, &record.EventID, &record.EnqueueOrdinal); err != nil { return false, 0, nil, err } records = append(records, record) } if err := rows.Err(); err != nil { return false, 0, nil, err } ok, violations, samples := evaluateLaneOrder(records) return ok, violations, samples, nil } func evaluateLaneOrder(records []laneOrderRecord) (bool, int, []string) { lastByLane := map[string]int64{} violations := 0 samples := []string{} for _, record := range records { if record.EnqueueOrdinal <= 0 { continue } last := lastByLane[record.Lane] if record.EnqueueOrdinal <= last { violations++ if len(samples) < 10 { samples = append(samples, fmt.Sprintf("%s event %s enqueue_ordinal %d after %d (%s)", record.Lane, record.EventID, record.EnqueueOrdinal, last, record.WorkID)) } continue } lastByLane[record.Lane] = record.EnqueueOrdinal } return violations == 0, violations, samples } func explainClaimSelect(ctx context.Context, conn *sql.Conn, opts runOptions, laneOrdinalAvailable bool) ([]string, bool, error) { now := time.Now().UTC() query := ` EXPLAIN SELECT w.WorkId FROM _System_ConnectorWorkItem w WHERE w.InstalledPackId = ? AND w.Generation = ? AND w.State IN ('pending', 'retry') AND w.RunAfter <= ? ORDER BY w.RunAfter ASC, w.WorkId ASC LIMIT 1024` args := []any{opts.PackID, opts.Generation, now} if laneOrdinalAvailable { query = ` EXPLAIN SELECT w.WorkId FROM _System_ConnectorWorkItem w WHERE w.InstalledPackId = ? AND w.Generation = ? AND w.State IN ('pending', 'retry') AND w.RunAfter <= ? AND NOT EXISTS ( SELECT 1 FROM _System_ConnectorWorkItem prior WHERE prior.InstalledPackId = w.InstalledPackId AND prior.Generation = w.Generation AND prior.State IN ('pending', 'retry') AND prior.RunAfter <= ? AND COALESCE(prior.SourceLaneKey, '') = COALESCE(w.SourceLaneKey, '') AND ( prior.LaneOrdinal < w.LaneOrdinal OR (prior.LaneOrdinal = w.LaneOrdinal AND prior.RunAfter < w.RunAfter) OR (prior.LaneOrdinal = w.LaneOrdinal AND prior.RunAfter = w.RunAfter AND prior.WorkId < w.WorkId) ) ) ORDER BY w.LaneOrdinal ASC, w.RunAfter ASC, w.WorkId ASC LIMIT 1024` args = []any{opts.PackID, opts.Generation, now, now} } rows, err := conn.QueryContext(ctx, query, args...) if err != nil { return nil, false, err } defer rows.Close() cols, err := rows.Columns() if err != nil { return nil, false, err } values := make([]sql.NullString, len(cols)) ptrs := make([]any, len(cols)) for i := range values { ptrs[i] = &values[i] } var summary []string broad := false for rows.Next() { if err := rows.Scan(ptrs...); err != nil { return nil, false, err } parts := make([]string, 0, len(cols)) for i, col := range cols { if !values[i].Valid { continue } value := values[i].String if col == "id" || col == "task" || col == "operator info" { parts = append(parts, col+"="+value) } if strings.Contains(value, "TableFullScan") || strings.Contains(value, "IndexFullScan") { broad = true } } summary = append(summary, strings.Join(parts, " | ")) } return summary, broad, rows.Err() } func countScalar(ctx context.Context, conn *sql.Conn, query string, args ...any) (int, error) { var count int if err := conn.QueryRowContext(ctx, query, args...).Scan(&count); err != nil { return 0, err } return count, nil } func redactDSN(dsn string) string { if strings.Contains(dsn, "127.0.0.1:4000") { return "root@tcp(127.0.0.1:4000)/..." } if i := strings.Index(dsn, "@"); i >= 0 { return "[redacted]" + dsn[i:] } return "[redacted]" } func runGitOrEmpty(args ...string) string { out, err := exec.Command("git", args...).Output() if err != nil { return "" } return strings.TrimSpace(string(out)) }