Merged
Fence obsolete repository-document source callbacks #172
urandonAdd on 27 days ago
Fence obsolete repository-document source callbacks #172
Merged
Merged fromcodex/issue-188-source-callback-authorityintomain
Total: 7 files changed+289-32
| @@ -99,6 +99,15 @@ gitcode-mcp repo-docs rebind --repo owner/repo --registration-id REG \ | |||
| 99 | --source-registration-generation 1 --repository-path /new/worktree | 99 | --source-registration-generation 1 --repository-path /new/worktree |
| 100 | ``` | 100 | ``` |
| 101 | 101 | ||
| 102 | +Rebind fences the previous source generation. A late reconcile completion or | ||
| 103 | +scheduling failure from that old registration/source/path cannot change the | ||
| 104 | +replacement's state, inspection history, retry/error state, active-job links or | ||
| 105 | +polling timestamps. It returns the current sanitized authority without writing | ||
| 106 | +the registry. Only a matching active job intent can mark the current source | ||
| 107 | +`indexing`; current-generation failures still apply normally. Registry restart | ||
| 108 | +preserves that authority through the existing exact source-id redirects. This | ||
| 109 | +is shared by CLI, MCP and Admin status, not an observation-side repair. | ||
| 110 | + | ||
| 102 | Public status exposes only opaque Git | 111 | Public status exposes only opaque Git |
| 103 | store/worktree refs. The coordinator then polls committed `HEAD` and policy | 112 | store/worktree refs. The coordinator then polls committed `HEAD` and policy |
| 104 | identity every minute and coalesces a new immutable-set job when either | 113 | identity every minute and coalesces a new immutable-set job when either |
| @@ -167,10 +167,14 @@ survive strict replay; paths, source ids, content and raw safe-looking error | |||
| 167 | classes remain excluded. Registration and source captures share one bounded | 167 | classes remain excluded. Registration and source captures share one bounded |
| 168 | operation-local buffer and publication boundary. | 168 | operation-local buffer and publication boundary. |
| 169 | Actual queued-job/rebind/fence and reconcile-snapshot/late-scheduling-failure | 169 | Actual queued-job/rebind/fence and reconcile-snapshot/late-scheduling-failure |
| 170 | -fixtures pair obsolete owners with current-generation controls. They preserve | 170 | +fixtures pair obsolete owners with current-generation controls. Obsolete |
| 171 | -existing execution results while requiring exactly one count-only loss for an | 171 | +completion and failure callbacks must preserve the entire replacement authority |
| 172 | -obsolete changed pair, one confirmed source capture for a current pair, and no | 172 | +and scheduling state, perform zero registry writes, and manufacture neither a |
| 173 | -extra boundary on unchanged replay. The held generation-fence test waits for an | 173 | +transition nor diagnostic loss. Current controls still apply normally. Restart |
| 174 | +readback qualifies the exact existing source redirect and retains every other | ||
| 175 | +source field. A deterministic registration/cache/repository/source/path/profile | ||
| 176 | +mismatch matrix and active-job intent controls fence unrelated work. The held | ||
| 177 | +generation-fence test waits for an | ||
| 174 | accepted FIFO collector marker after registry save/unlock before asserting no | 178 | accepted FIFO collector marker after registry save/unlock before asserting no |
| 175 | early maintenance publication; releasing the real fence conserves the transition. | 179 | early maintenance publication; releasing the real fence conserves the transition. |
| 176 | Identity-resolution fixtures apply real legacy registry conflicts and physical | 180 | Identity-resolution fixtures apply real legacy registry conflicts and physical |
| @@ -2826,16 +2826,27 @@ func repositoryDocsRetryReady(state *RepositoryDocsMaintenanceState, now time.Ti | |||
| 2826 | 2826 | ||
| 2827 | func (m *MaintenanceManager) finishReconcileEntry(ctx context.Context, registrationID string, snapshot MaintenanceEntry, contentGeneration, covered int64, ragStatus, namespaceID string, frontiers []cache.MaintenanceFrontier, activeSync, activeRAG, activeRepositoryDocs Job, now time.Time) MaintenanceEntry { | 2827 | func (m *MaintenanceManager) finishReconcileEntry(ctx context.Context, registrationID string, snapshot MaintenanceEntry, contentGeneration, covered int64, ragStatus, namespaceID string, frontiers []cache.MaintenanceFrontier, activeSync, activeRAG, activeRepositoryDocs Job, now time.Time) MaintenanceEntry { |
| 2828 | m.mu.Lock() | 2828 | m.mu.Lock() |
| 2829 | + defer m.mu.Unlock() | ||
| 2829 | entry := m.entries[registrationID] | 2830 | entry := m.entries[registrationID] |
| 2831 | + if entry == nil { | ||
| 2832 | + return MaintenanceEntry{} | ||
| 2833 | + } | ||
| 2834 | + _, currentEntry := m.reconcileAuthorityLocked(&snapshot) | ||
| 2835 | + _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&snapshot) | ||
| 2836 | + if registrationID != snapshot.RegistrationID || !currentEntry || snapshot.RepositoryDocs != nil && !currentSourceAuthority { | ||
| 2837 | + return cloneMaintenanceEntry(entry) | ||
| 2838 | + } | ||
| 2839 | + if activeRepositoryDocs.ID != "" && !repositoryDocsJobMatchesSnapshot(activeRepositoryDocs, snapshot) { | ||
| 2840 | + activeRepositoryDocs = Job{} | ||
| 2841 | + } | ||
| 2830 | beforeObservation := maintenanceObservedState(entry) | 2842 | beforeObservation := maintenanceObservedState(entry) |
| 2831 | - observedSource := m.sources[registrationID][snapshot.repositorySourceID] | 2843 | + observedSource := authoritySource |
| 2832 | beforeSourceObservation := maintenanceObservedSourceState(observedSource) | 2844 | beforeSourceObservation := maintenanceObservedSourceState(observedSource) |
| 2833 | entry.ContentGeneration = contentGeneration | 2845 | entry.ContentGeneration = contentGeneration |
| 2834 | entry.CoveredGeneration = covered | 2846 | entry.CoveredGeneration = covered |
| 2835 | entry.RAGStatus = ragStatus | 2847 | entry.RAGStatus = ragStatus |
| 2836 | entry.SyncStage = snapshot.SyncStage | 2848 | entry.SyncStage = snapshot.SyncStage |
| 2837 | entry.RAGStage = snapshot.RAGStage | 2849 | entry.RAGStage = snapshot.RAGStage |
| 2838 | - _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&snapshot) | ||
| 2839 | if currentSourceAuthority { | 2850 | if currentSourceAuthority { |
| 2840 | // Inspection is separately confirmed at its actual completion. A later | 2851 | // Inspection is separately confirmed at its actual completion. A later |
| 2841 | // stale finish must not overwrite that transaction or bypass its failure. | 2852 | // stale finish must not overwrite that transaction or bypass its failure. |
| @@ -2875,13 +2886,16 @@ func (m *MaintenanceManager) finishReconcileEntry(ctx context.Context, registrat | |||
| 2875 | entry.NextReconcileAt = now.Add(time.Minute) | 2886 | entry.NextReconcileAt = now.Add(time.Minute) |
| 2876 | persisted := m.saveLocked() == nil | 2887 | persisted := m.saveLocked() == nil |
| 2877 | captureMaintenanceObservation(ctx, beforeObservation, entry, persisted) | 2888 | captureMaintenanceObservation(ctx, beforeObservation, entry, persisted) |
| 2878 | - // An obsolete callback may retain an existing execution-state mutation. Do | 2889 | + captureMaintenanceSourceObservation(ctx, beforeSourceObservation, entry, observedSource, persisted && currentSourceAuthority) |
| 2879 | - // not certify it as a current source observation: conserve the changed pair | 2890 | + return cloneMaintenanceEntry(entry) |
| 2880 | - // as unconfirmed loss without altering execution or persistence behavior. | 2891 | +} |
| 2881 | - captureMaintenanceSourceObservation(ctx, beforeSourceObservation, entry, observedSource, persisted && currentSourceAuthority && observedSource == authoritySource) | 2892 | + |
| 2882 | - updated := cloneMaintenanceEntry(entry) | 2893 | +func repositoryDocsJobMatchesSnapshot(job Job, snapshot MaintenanceEntry) bool { |
| 2883 | - m.mu.Unlock() | 2894 | + state := snapshot.RepositoryDocs |
| 2884 | - return updated | 2895 | + return state != nil && job.Type == RepositoryDocsIndexJobType && jobActiveStatus(job.Status) && |
| 2896 | + job.CacheUUID == snapshot.CacheUUID && job.RepoID == snapshot.RepoID && | ||
| 2897 | + job.RegistrationID == snapshot.RegistrationID && job.SourceRegistrationID == state.SourceRegistrationID && | ||
| 2898 | + job.SourceRegistrationGeneration == state.SourceRegistrationGeneration | ||
| 2885 | } | 2899 | } |
| 2886 | 2900 | ||
| 2887 | func (m *MaintenanceManager) reconcileRepositoryDocsState(ctx context.Context, store *cache.SQLiteStore, entry *MaintenanceEntry, now time.Time) bool { | 2901 | func (m *MaintenanceManager) reconcileRepositoryDocsState(ctx context.Context, store *cache.SQLiteStore, entry *MaintenanceEntry, now time.Time) bool { |
| @@ -2962,11 +2976,11 @@ func (m *MaintenanceManager) updateRepositoryDocsFailure(ctx context.Context, id | |||
| 2962 | return MaintenanceEntry{} | 2976 | return MaintenanceEntry{} |
| 2963 | } | 2977 | } |
| 2964 | source := m.sources[id][strings.TrimSpace(sourceID)] | 2978 | source := m.sources[id][strings.TrimSpace(sourceID)] |
| 2965 | - if source == nil { | 2979 | + _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&expected) |
| 2980 | + if id != expected.RegistrationID || !currentSourceAuthority || source == nil || source != authoritySource { | ||
| 2966 | return cloneMaintenanceEntry(entry) | 2981 | return cloneMaintenanceEntry(entry) |
| 2967 | } | 2982 | } |
| 2968 | beforeObservation := maintenanceObservedSourceState(source) | 2983 | beforeObservation := maintenanceObservedSourceState(source) |
| 2969 | - _, authoritySource, currentSourceAuthority := m.inspectionAuthorityLocked(&expected) | ||
| 2970 | source.State.State = "degraded" | 2984 | source.State.State = "degraded" |
| 2971 | source.State.LastErrorClass = class | 2985 | source.State.LastErrorClass = class |
| 2972 | source.State.LastError = message | 2986 | source.State.LastError = message |
| @@ -2976,7 +2990,7 @@ func (m *MaintenanceManager) updateRepositoryDocsFailure(ctx context.Context, id | |||
| 2976 | entry.LastReconciledAt = m.now() | 2990 | entry.LastReconciledAt = m.now() |
| 2977 | entry.NextReconcileAt = entry.LastReconciledAt.Add(time.Minute) | 2991 | entry.NextReconcileAt = entry.LastReconciledAt.Add(time.Minute) |
| 2978 | persisted := m.saveLocked() == nil | 2992 | persisted := m.saveLocked() == nil |
| 2979 | - captureMaintenanceSourceObservation(ctx, beforeObservation, entry, source, persisted && currentSourceAuthority && source == authoritySource) | 2993 | + captureMaintenanceSourceObservation(ctx, beforeObservation, entry, source, persisted) |
| 2980 | return cloneMaintenanceEntry(entry) | 2994 | return cloneMaintenanceEntry(entry) |
| 2981 | } | 2995 | } |
| 2982 | 2996 | ||
| @@ -112,7 +112,8 @@ func captureMaintenanceObservation(ctx context.Context, before maintenanceObserv | |||
| 112 | } | 112 | } |
| 113 | 113 | ||
| 114 | // Source confirmation requires both the existing save and current originating | 114 | // Source confirmation requires both the existing save and current originating |
| 115 | -// authority. Retained stale-owner changes are count-only loss, not certification. | 115 | +// authority. Failed retained writes are count-only loss, not certification; |
| 116 | +// rejected obsolete callbacks mutate nothing and contribute neither. | ||
| 116 | func captureMaintenanceSourceObservation(ctx context.Context, before maintenanceObservationState, entry *MaintenanceEntry, source *repositoryDocsRegisteredSource, confirmed bool) { | 117 | func captureMaintenanceSourceObservation(ctx context.Context, before maintenanceObservationState, entry *MaintenanceEntry, source *repositoryDocsRegisteredSource, confirmed bool) { |
| 117 | if source == nil { | 118 | if source == nil { |
| 118 | return | 119 | return |
| @@ -6,6 +6,7 @@ import ( | |||
| 6 | "errors" | 6 | "errors" |
| 7 | "os" | 7 | "os" |
| 8 | "path/filepath" | 8 | "path/filepath" |
| 9 | + "reflect" | ||
| 9 | "strings" | 10 | "strings" |
| 10 | "sync" | 11 | "sync" |
| 11 | "sync/atomic" | 12 | "sync/atomic" |
| @@ -107,7 +108,9 @@ func TestMaintenanceSourceObservationActualInspectionFailureRecovery(t *testing. | |||
| 107 | at = at.Add(time.Minute) | 108 | at = at.Add(time.Minute) |
| 108 | run(Job{}) | 109 | run(Job{}) |
| 109 | run(Job{}) // repeated successful inspection has no finite pair change | 110 | run(Job{}) // repeated successful inspection has no finite pair change |
| 110 | - run(Job{ID: "fixture-active-source"}) | 111 | + run(Job{ID: "fixture-active-source", Type: RepositoryDocsIndexJobType, Status: JobStatusQueued, |
| 112 | + CacheUUID: m.entries[reg].CacheUUID, RepoID: m.entries[reg].RepoID, RegistrationID: reg, | ||
| 113 | + SourceRegistrationID: id, SourceRegistrationGeneration: m.sources[reg][id].State.SourceRegistrationGeneration}) | ||
| 111 | run(Job{}) | 114 | run(Job{}) |
| 112 | if m.sources[reg][id].State.State != "not_indexed" || m.sources[reg][id].State.Inspection.Result != "succeeded" { | 115 | if m.sources[reg][id].State.State != "not_indexed" || m.sources[reg][id].State.Inspection.Result != "succeeded" { |
| 113 | t.Fatal("actual inspection/indexing owner changed authority") | 116 | t.Fatal("actual inspection/indexing owner changed authority") |
| @@ -312,10 +315,14 @@ func TestMaintenanceSourceObservationStaleSnapshotAndSharedOverflow(t *testing.T | |||
| 312 | t.Fatal("stale completion changed current source authority") | 315 | t.Fatal("stale completion changed current source authority") |
| 313 | } | 316 | } |
| 314 | buffer := ctx.Value(maintenanceObservationKey{}).(*maintenanceObservationBuffer) | 317 | buffer := ctx.Value(maintenanceObservationKey{}).(*maintenanceObservationBuffer) |
| 315 | - // Parent's actual enrolled->indexing is the only transition above. The stale | 318 | + if buffer.count != 0 || buffer.lost != 0 { |
| 316 | - // source's ready projection was never the current authority owner. | 319 | + t.Fatal("rejected stale source callback manufactured an event or loss") |
| 320 | + } | ||
| 321 | + // A qualified current finish supplies the real enrolled->indexing parent | ||
| 322 | + // transition; obsolete work can no longer mutate even the parent state. | ||
| 323 | + m.finishReconcileEntry(ctx, reg, cloneMaintenanceEntryPrivate(m.entries[reg]), 0, 0, "", "", nil, Job{}, Job{}, Job{}, time.Now().UTC()) | ||
| 317 | if buffer.count != 1 || buffer.captures[0].scope != "registration" { | 324 | if buffer.count != 1 || buffer.captures[0].scope != "registration" { |
| 318 | - t.Fatal("stale source completion manufactured an event") | 325 | + t.Fatal("current completion lost parent observation") |
| 319 | } | 326 | } |
| 320 | entry, source := m.entries[reg], m.sources[reg][id] | 327 | entry, source := m.entries[reg], m.sources[reg][id] |
| 321 | for i := 0; i < maxMaintenanceObservations+5; i++ { | 328 | for i := 0; i < maxMaintenanceObservations+5; i++ { |
| @@ -366,17 +373,32 @@ func TestMaintenanceSourceObservationActualObsoleteJobAndCurrentControl(t *testi | |||
| 366 | } | 373 | } |
| 367 | c := maintenanceEventCollector(t) | 374 | c := maintenanceEventCollector(t) |
| 368 | m.jobs.SetObservation(c) | 375 | m.jobs.SetObservation(c) |
| 376 | + before := cloneMaintenanceEntryPrivate(m.entries[reg]) | ||
| 377 | + beforeSource := *m.sources[reg][id] | ||
| 378 | + writes := 0 | ||
| 379 | + m.writeFile = func(name string, data []byte, permissions os.FileMode) error { | ||
| 380 | + writes++ | ||
| 381 | + return durableAtomicWriteFile(name, data, permissions) | ||
| 382 | + } | ||
| 369 | ctx, flush := beginMaintenanceObservation(context.Background(), m.jobs) | 383 | ctx, flush := beginMaintenanceObservation(context.Background(), m.jobs) |
| 370 | defer flush() | 384 | defer flush() |
| 371 | updated := m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC()) | 385 | updated := m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC()) |
| 372 | - // Execution semantics predate instrumentation and remain unchanged. | 386 | + if stale { |
| 373 | - if updated.State != "indexing" || m.sources[reg][id].State.State != "indexing" || m.sources[reg][id].State.SourceRegistrationGeneration != 2 { | 387 | + if !reflect.DeepEqual(updated, cloneMaintenanceEntry(&before)) { |
| 374 | - t.Fatal("observation changed retained execution behavior") | 388 | + t.Fatal("obsolete completion returned stale state") |
| 389 | + } | ||
| 390 | + assertObsoleteSourceCallbackUnchanged(t, m, reg, id, before, beforeSource, writes) | ||
| 391 | + } else if updated.State != "indexing" || m.sources[reg][id].State.State != "indexing" || m.sources[reg][id].State.SourceRegistrationGeneration != 2 || writes == 0 { | ||
| 392 | + t.Fatal("current source completion did not apply") | ||
| 375 | } | 393 | } |
| 376 | assertMaintenanceSourceAuthorityCaptures(t, ctx, stale) | 394 | assertMaintenanceSourceAuthorityCaptures(t, ctx, stale) |
| 377 | // Unchanged replay must not retroactively certify or recount the pair. | 395 | // Unchanged replay must not retroactively certify or recount the pair. |
| 378 | m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC()) | 396 | m.finishReconcileEntry(ctx, reg, snapshot, 0, 0, "", "", nil, Job{}, Job{}, active, time.Now().UTC()) |
| 379 | assertMaintenanceSourceAuthorityCaptures(t, ctx, stale) | 397 | assertMaintenanceSourceAuthorityCaptures(t, ctx, stale) |
| 398 | + if stale { | ||
| 399 | + assertObsoleteSourceCallbackUnchanged(t, m, reg, id, before, beforeSource, writes) | ||
| 400 | + } | ||
| 401 | + assertSourceAuthorityRestart(t, m, reg, id, *m.sources[reg][id]) | ||
| 380 | }) | 402 | }) |
| 381 | } | 403 | } |
| 382 | } | 404 | } |
| @@ -420,19 +442,33 @@ func TestMaintenanceSourceObservationActualLateFailureAndCurrentControl(t *testi | |||
| 420 | t.Fatal("actual source rebind unavailable") | 442 | t.Fatal("actual source rebind unavailable") |
| 421 | } | 443 | } |
| 422 | } | 444 | } |
| 445 | + before := cloneMaintenanceEntryPrivate(m.entries[reg]) | ||
| 446 | + beforeSource := *m.sources[reg][id] | ||
| 447 | + var writes atomic.Int64 | ||
| 448 | + m.writeFile = func(name string, data []byte, permissions os.FileMode) error { | ||
| 449 | + writes.Add(1) | ||
| 450 | + return durableAtomicWriteFile(name, data, permissions) | ||
| 451 | + } | ||
| 423 | unblock() | 452 | unblock() |
| 424 | select { | 453 | select { |
| 425 | case updated := <-done: | 454 | case updated := <-done: |
| 426 | - if updated.RepositoryDocs == nil || updated.RepositoryDocs.State != "degraded" { | 455 | + if stale { |
| 427 | - t.Fatal("observation changed existing failure return") | 456 | + if !reflect.DeepEqual(updated, cloneMaintenanceEntry(&before)) { |
| 457 | + t.Fatal("obsolete failure returned stale state") | ||
| 458 | + } | ||
| 459 | + } else if updated.RepositoryDocs == nil || updated.RepositoryDocs.State != "degraded" { | ||
| 460 | + t.Fatal("current source failure did not apply") | ||
| 428 | } | 461 | } |
| 429 | case <-time.After(5 * time.Second): | 462 | case <-time.After(5 * time.Second): |
| 430 | t.Fatal("actual source scheduling failure did not unwind") | 463 | t.Fatal("actual source scheduling failure did not unwind") |
| 431 | } | 464 | } |
| 432 | - if m.sources[reg][id].State.State != "degraded" || m.sources[reg][id].State.LastErrorClass != "repository_docs_provider_boundary_blocked" { | 465 | + if stale { |
| 466 | + assertObsoleteSourceCallbackUnchanged(t, m, reg, id, before, beforeSource, int(writes.Load())) | ||
| 467 | + } else if m.sources[reg][id].State.State != "degraded" || m.sources[reg][id].State.LastErrorClass != "repository_docs_provider_boundary_blocked" || writes.Load() == 0 { | ||
| 433 | t.Fatal("actual local-only scheduling boundary unavailable") | 468 | t.Fatal("actual local-only scheduling boundary unavailable") |
| 434 | } | 469 | } |
| 435 | assertMaintenanceSourceAuthorityCaptures(t, ctx, stale) | 470 | assertMaintenanceSourceAuthorityCaptures(t, ctx, stale) |
| 471 | + assertSourceAuthorityRestart(t, m, reg, id, *m.sources[reg][id]) | ||
| 436 | }) | 472 | }) |
| 437 | } | 473 | } |
| 438 | } | 474 | } |
| @@ -448,10 +484,54 @@ func assertMaintenanceSourceAuthorityCaptures(t *testing.T, ctx context.Context, | |||
| 448 | } | 484 | } |
| 449 | expected, lost := 1, uint64(0) | 485 | expected, lost := 1, uint64(0) |
| 450 | if stale { | 486 | if stale { |
| 451 | - expected, lost = 0, 1 | 487 | + expected = 0 |
| 452 | } | 488 | } |
| 453 | if sourceCaptures != expected || buffer.lost != lost { | 489 | if sourceCaptures != expected || buffer.lost != lost { |
| 454 | - t.Fatal("current confirmation and obsolete count-only loss not conserved") | 490 | + t.Fatal("source confirmation or unchanged obsolete callback accounting incorrect") |
| 491 | + } | ||
| 492 | +} | ||
| 493 | + | ||
| 494 | +func assertObsoleteSourceCallbackUnchanged(t *testing.T, m *MaintenanceManager, reg, id string, before MaintenanceEntry, beforeSource repositoryDocsRegisteredSource, writes int) { | ||
| 495 | + t.Helper() | ||
| 496 | + if writes != 0 || !reflect.DeepEqual(cloneMaintenanceEntryPrivate(m.entries[reg]), before) || *m.sources[reg][id] != beforeSource { | ||
| 497 | + t.Fatal("obsolete callback changed replacement authority or persisted state") | ||
| 498 | + } | ||
| 499 | + list, err := m.List(context.Background()) | ||
| 500 | + if err != nil || len(list.Entries) != 1 || !reflect.DeepEqual(list.Entries[0], cloneMaintenanceEntry(&before)) { | ||
| 501 | + t.Fatal("public maintenance projection changed after obsolete callback") | ||
| 502 | + } | ||
| 503 | +} | ||
| 504 | + | ||
| 505 | +func assertSourceAuthorityRestart(t *testing.T, m *MaintenanceManager, reg, id string, want repositoryDocsRegisteredSource) { | ||
| 506 | + t.Helper() | ||
| 507 | + reloaded := NewMaintenanceManager(m.manager, NewJobManager(""), m.path) | ||
| 508 | + if reloaded.Load() != nil { | ||
| 509 | + t.Fatal("durable source authority changed across restart") | ||
| 510 | + } | ||
| 511 | + // Existing registry canonicalization may redirect a rebound source's | ||
| 512 | + // original id to the id of its replacement Git/worktree/profile authority. | ||
| 513 | + // Require that exact redirect, not an arbitrary substitute source. | ||
| 514 | + canonicalID := reloaded.resolveSourceRegistrationIDLocked(id) | ||
| 515 | + if canonicalID != repositoryDocsSourceRegistrationID(reg, want.State.GitStoreRef, want.State.WorktreeRef, want.Profile) || reloaded.sources[reg][canonicalID] == nil { | ||
| 516 | + t.Fatal("durable source redirect does not match replacement authority") | ||
| 517 | + } | ||
| 518 | + want.State.SourceRegistrationID = canonicalID | ||
| 519 | + got := *reloaded.sources[reg][canonicalID] | ||
| 520 | + if got != want { | ||
| 521 | + changed := []string{} | ||
| 522 | + if got.RepositoryPath != want.RepositoryPath { | ||
| 523 | + changed = append(changed, "path") | ||
| 524 | + } | ||
| 525 | + if got.Profile != want.Profile { | ||
| 526 | + changed = append(changed, "profile") | ||
| 527 | + } | ||
| 528 | + a, b := reflect.ValueOf(got.State), reflect.ValueOf(want.State) | ||
| 529 | + for i := 0; i < a.NumField(); i++ { | ||
| 530 | + if !reflect.DeepEqual(a.Field(i).Interface(), b.Field(i).Interface()) { | ||
| 531 | + changed = append(changed, a.Type().Field(i).Name) | ||
| 532 | + } | ||
| 533 | + } | ||
| 534 | + t.Fatalf("durable source authority changed across restart: fields %s", strings.Join(changed, ",")) | ||
| 455 | } | 535 | } |
| 456 | } | 536 | } |
| 457 | 537 | ||
| @@ -56,16 +56,29 @@ func (m *MaintenanceManager) retainRepositoryDocsInspection(snapshot *Maintenanc | |||
| 56 | } | 56 | } |
| 57 | 57 | ||
| 58 | func (m *MaintenanceManager) inspectionAuthorityLocked(snapshot *MaintenanceEntry) (*MaintenanceEntry, *repositoryDocsRegisteredSource, bool) { | 58 | func (m *MaintenanceManager) inspectionAuthorityLocked(snapshot *MaintenanceEntry) (*MaintenanceEntry, *repositoryDocsRegisteredSource, bool) { |
| 59 | - entry := m.entries[snapshot.RegistrationID] | 59 | + entry, currentEntry := m.reconcileAuthorityLocked(snapshot) |
| 60 | state := snapshot.RepositoryDocs | 60 | state := snapshot.RepositoryDocs |
| 61 | - if entry == nil || state == nil { | 61 | + if !currentEntry || state == nil { |
| 62 | return entry, nil, false | 62 | return entry, nil, false |
| 63 | } | 63 | } |
| 64 | source := m.sources[snapshot.RegistrationID][state.SourceRegistrationID] | 64 | source := m.sources[snapshot.RegistrationID][state.SourceRegistrationID] |
| 65 | - current := source != nil && entry.Generation == snapshot.Generation && entry.CacheUUID == snapshot.CacheUUID && entry.RepoID == snapshot.RepoID && entry.PathFingerprint == snapshot.PathFingerprint && source.RepositoryPath == snapshot.repositoryPath && source.State.SourceRegistrationGeneration == state.SourceRegistrationGeneration | 65 | + current := source != nil && state.SourceRegistrationID == snapshot.repositorySourceID && |
| 66 | + source.State.SourceRegistrationID == state.SourceRegistrationID && | ||
| 67 | + source.RepositoryPath == snapshot.repositoryPath && source.Profile == snapshot.repositoryProfile && | ||
| 68 | + source.State.SourceRegistrationGeneration == state.SourceRegistrationGeneration | ||
| 66 | return entry, source, current | 69 | return entry, source, current |
| 67 | } | 70 | } |
| 68 | 71 | ||
| 72 | +// Completion authority belongs to the snapshot that started the work, not a | ||
| 73 | +// replacement registration found later under the same map key. Call under mu | ||
| 74 | +// before mutating either scheduling state or source state. | ||
| 75 | +func (m *MaintenanceManager) reconcileAuthorityLocked(snapshot *MaintenanceEntry) (*MaintenanceEntry, bool) { | ||
| 76 | + entry := m.entries[snapshot.RegistrationID] | ||
| 77 | + return entry, entry != nil && entry.RegistrationID == snapshot.RegistrationID && | ||
| 78 | + entry.Generation == snapshot.Generation && entry.CacheUUID == snapshot.CacheUUID && | ||
| 79 | + entry.RepoID == snapshot.RepoID && entry.PathFingerprint == snapshot.PathFingerprint | ||
| 80 | +} | ||
| 81 | + | ||
| 69 | func inspectionHealthReason(class string) string { | 82 | func inspectionHealthReason(class string) string { |
| 70 | switch class { | 83 | switch class { |
| 71 | case "git_store": | 84 | case "git_store": |