From 59cdf871bb163641695aac38755ed7d4d569600b Mon Sep 17 00:00:00 2001 From: Mark Sujew Date: Wed, 26 Aug 2026 16:43:14 +0200 Subject: [PATCH] Support per-document read lock --- cmd/fastbelt/generate.go | 2 +- document.go | 31 +- examples/statemachine/benchmark_test.go | 6 +- examples/statemachine/builder_reset_test.go | 2 +- server/completion_contributor.go | 3 + server/completion_provider.go | 11 + server/completion_snippet.go | 3 + server/completion_triggers.go | 3 + server/definition_provider.go | 9 + server/diagnostics_publisher.go | 3 + server/document_highlight_provider.go | 9 + server/document_symbol_provider.go | 8 + server/folding_range_provider.go | 9 + server/hover_provider.go | 9 + server/references_finder.go | 3 + server/references_provider.go | 10 + server/rename_provider.go | 10 + server/server.go | 55 +- server/workspace_symbol_provider.go | 9 + test/doc_fixture.go | 4 +- test/doc_fixture_lsp.go | 2 +- test/fixture.go | 4 +- workspace/builder.go | 93 +- workspace/doc.go | 47 +- workspace/document_updater.go | 6 +- workspace/lock.go | 207 +++-- workspace/lock_test.go | 888 +++++++++++--------- workspace/services.go | 2 +- 28 files changed, 925 insertions(+), 523 deletions(-) diff --git a/cmd/fastbelt/generate.go b/cmd/fastbelt/generate.go index 78b399de..ad2b6783 100644 --- a/cmd/fastbelt/generate.go +++ b/cmd/fastbelt/generate.go @@ -67,7 +67,7 @@ func runGenerateCLI(opts generateOptions) error { if err != nil { return err } - if err := builder.Build(context.Background(), []*core.Document{document}, nil); err != nil { + if err := builder.Build(context.Background(), []*core.Document{document}); err != nil { return err } diff --git a/document.go b/document.go index d5b51075..5e59152e 100644 --- a/document.go +++ b/document.go @@ -7,6 +7,7 @@ package fastbelt import ( "strings" "sync" + "sync/atomic" "typefox.dev/fastbelt/textdoc" "typefox.dev/lsp" @@ -17,12 +18,15 @@ import ( // For example, the Root node may be nil if the document has not been parsed yet. // // Access to the fields of Document should be synchronized using a [typefox.dev/fastbelt/workspace] Lock. +// Readers admitted based on the document's build state (workspace Lock.ReadAt) may only access +// the fields produced by the states they requested; later fields may still be under construction. // The document struct should never be copied after creation. type Document struct { // URI identifies the document in the workspace. URI URI - // State tracks which build phases already ran for this document. - State DocumentState + // state tracks which build phases already ran for this document. + // Access it through [Document.State] and [Document.SetState]. + state atomic.Uint32 // Root is the AST root produced by parsing. // It is nil until parsing succeeds. Root AstNode @@ -78,7 +82,6 @@ func NewDocument(textDoc textdoc.Handle) *Document { uri := ParseURI(string(textDoc.URI())) return &Document{ URI: uri, - State: 0, TextDoc: textDoc, Root: nil, LocalSymbols: nil, @@ -108,6 +111,24 @@ func NewDocumentFromString(uri, languageId, content string) (*Document, error) { return doc, nil } +// State returns the build phases that have completed for this document. +// +// It is safe to call concurrently with a running build: the atomic load pairs +// with [Document.SetState], so a caller that observes a state bit is +// guaranteed to see all document data written by that build phase. +func (d *Document) State() DocumentState { + return DocumentState(d.state.Load()) +} + +// SetState replaces the document's build state. +// +// Callers must hold the workspace write lock and must write all document data +// belonging to a phase before setting its bit; the atomic store publishes that +// data to concurrent [Document.State] readers. +func (d *Document) SetState(state DocumentState) { + d.state.Store(uint32(state)) +} + // DocumentState is a bitmask capturing the already completed build phases of a document. type DocumentState uint32 @@ -152,9 +173,9 @@ func (s DocumentState) String() string { return strings.Join(flags, " | ") } -// Has reports whether flag is set in s. +// Has reports whether flag is set in s. If flag is a combination of multiple bits, Has reports whether all of them are set. func (s DocumentState) Has(flag DocumentState) bool { - return s&flag != 0 + return s&flag == flag } // With returns s with flag set. diff --git a/examples/statemachine/benchmark_test.go b/examples/statemachine/benchmark_test.go index 5cdf559e..e0aaa705 100644 --- a/examples/statemachine/benchmark_test.go +++ b/examples/statemachine/benchmark_test.go @@ -54,8 +54,8 @@ func BenchmarkWorkspaceCycle(b *testing.B) { } start := time.Now() - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - if err := builder.Build(ctx, docs, downgrade); err != nil { + lock.Write(context.Background(), func(ctx context.Context) { + if err := builder.Build(ctx, docs); err != nil { b.Errorf("build failed: %v", err) } }) @@ -181,7 +181,7 @@ func BenchmarkLocalLinking(b *testing.B) { docs := []*fastbelt.Document{doc} builder := service.MustGet[workspace.Builder](srv) // Prebuild the file - we will reset the references later - if err := builder.Build(b.Context(), docs, nil); err != nil { + if err := builder.Build(b.Context(), docs); err != nil { b.Errorf("build failed: %v", err) } b.SetBytes(int64(len(content))) diff --git a/examples/statemachine/builder_reset_test.go b/examples/statemachine/builder_reset_test.go index 774fc3d7..d775da00 100644 --- a/examples/statemachine/builder_reset_test.go +++ b/examples/statemachine/builder_reset_test.go @@ -48,7 +48,7 @@ func TestResetKeepsLocalSymbols(t *testing.T) { t.Error("Reset kept ImportedSymbols although DocStateImportedSymbols was dropped") } - if err := builder.Build(f.Ctx(), []*core.Document{doc.Document}, nil); err != nil { + if err := builder.Build(f.Ctx(), []*core.Document{doc.Document}); err != nil { t.Fatalf("rebuild after reset failed: %v", err) } for _, ref := range doc.Document.References { diff --git a/server/completion_contributor.go b/server/completion_contributor.go index b0bb92d7..e56b1189 100644 --- a/server/completion_contributor.go +++ b/server/completion_contributor.go @@ -97,6 +97,9 @@ func NewDefaultCompletionContributor() CompletionContributor { return &DefaultCompletionContributor{} } +// Ensure DefaultCompletionContributor implements CompletionContributor. +var _ CompletionContributor = (*DefaultCompletionContributor)(nil) + // CompletionForToken emits a default item only for keyword tokens that contains a letter or digit. // This prevents trivial punctuation tokens from appearing in the default completion list. func (*DefaultCompletionContributor) CompletionForToken(_ context.Context, tt *core.TokenType, _ int, _ ContributorContext, accept CompletionAcceptor) { diff --git a/server/completion_provider.go b/server/completion_provider.go index ef80c2fb..9735c76d 100644 --- a/server/completion_provider.go +++ b/server/completion_provider.go @@ -42,6 +42,17 @@ func NewDefaultCompletionProvider(sc *service.Container) CompletionProvider { return &DefaultCompletionProvider{sc: sc} } +// The default required state for completion is DocStateLinked, which has to be +// satisfied across the whole workspace, because cross-references might require +// complex scoping and linking to be fully resolved. +func (s *DefaultCompletionProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateLinked, true +} + +// Ensure that the default completion provider correctly implements the interfaces +var _ DocumentStateRequirements = (*DefaultCompletionProvider)(nil) +var _ CompletionProvider = (*DefaultCompletionProvider)(nil) + // HandleCompletionRequest fulfils textDocument/completion. The flow is: // // 1. resolve doc + cursor offset; diff --git a/server/completion_snippet.go b/server/completion_snippet.go index bffb5ca8..1f870b28 100644 --- a/server/completion_snippet.go +++ b/server/completion_snippet.go @@ -55,6 +55,9 @@ type DefaultSnippetRegistry struct { snippets []SnippetTemplate } +// Ensure DefaultSnippetRegistry implements SnippetRegistry. +var _ SnippetRegistry = (*DefaultSnippetRegistry)(nil) + // NewDefaultSnippetRegistry returns an empty registry. func NewDefaultSnippetRegistry() SnippetRegistry { return &DefaultSnippetRegistry{} diff --git a/server/completion_triggers.go b/server/completion_triggers.go index 84b6f7a5..1e6f433a 100644 --- a/server/completion_triggers.go +++ b/server/completion_triggers.go @@ -18,6 +18,9 @@ type CompletionTriggers interface { // DefaultCompletionTriggers returns nil - no auto-open characters. type DefaultCompletionTriggers struct{} +// Ensure DefaultCompletionTriggers implements CompletionTriggers. +var _ CompletionTriggers = (*DefaultCompletionTriggers)(nil) + // NewDefaultCompletionTriggers returns the no-op trigger set. func NewDefaultCompletionTriggers() CompletionTriggers { return &DefaultCompletionTriggers{} diff --git a/server/definition_provider.go b/server/definition_provider.go index d80c95a2..f68ff6c9 100644 --- a/server/definition_provider.go +++ b/server/definition_provider.go @@ -25,10 +25,19 @@ type DefaultDefinitionProvider struct { sc *service.Container } +// Ensure the DefaultDefinitionProvider implements the expected interfaces. +var _ DefinitionProvider = (*DefaultDefinitionProvider)(nil) +var _ DocumentStateRequirements = (*DefaultDefinitionProvider)(nil) + func NewDefaultDefinitionProvider(sc *service.Container) DefinitionProvider { return &DefaultDefinitionProvider{sc: sc} } +// The default definition provider requires the document in question to be linked. +func (s *DefaultDefinitionProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateLinked, false +} + func (s *DefaultDefinitionProvider) HandleDefinitionRequest(ctx context.Context, params *lsp.DefinitionParams) ([]lsp.DefinitionLink, error) { documentManager := service.MustGet[workspace.DocumentManager](s.sc) uri := core.ParseURI(string(params.TextDocument.URI)) diff --git a/server/diagnostics_publisher.go b/server/diagnostics_publisher.go index 0c139464..c88d9558 100644 --- a/server/diagnostics_publisher.go +++ b/server/diagnostics_publisher.go @@ -27,6 +27,9 @@ type DiagnosticsPublisher struct { sc *service.Container } +// Ensure DiagnosticsPublisher implements InitializeParticipant. +var _ InitializeParticipant = (*DiagnosticsPublisher)(nil) + // NewDiagnosticsPublisher creates a new instance of [DiagnosticsPublisher]. func NewDiagnosticsPublisher(sc *service.Container) *DiagnosticsPublisher { return &DiagnosticsPublisher{sc: sc} diff --git a/server/document_highlight_provider.go b/server/document_highlight_provider.go index 84b187d8..19258566 100644 --- a/server/document_highlight_provider.go +++ b/server/document_highlight_provider.go @@ -17,6 +17,10 @@ type DocumentHighlightProvider interface { HandleDocumentHighlightRequest(ctx context.Context, params *lsp.DocumentHighlightParams) ([]lsp.DocumentHighlight, error) } +// Ensure DefaultDocumentHighlightProvider implements the expected interfaces. +var _ DocumentHighlightProvider = (*DefaultDocumentHighlightProvider)(nil) +var _ DocumentStateRequirements = (*DefaultDocumentHighlightProvider)(nil) + type DefaultDocumentHighlightProvider struct { sc *service.Container } @@ -25,6 +29,11 @@ func NewDefaultDocumentHighlightProvider(sc *service.Container) DocumentHighligh return &DefaultDocumentHighlightProvider{sc: sc} } +// The default document highlight provider requires the document in question to be linked. +func (s *DefaultDocumentHighlightProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateLinked, false +} + func (s *DefaultDocumentHighlightProvider) HandleDocumentHighlightRequest(ctx context.Context, params *lsp.DocumentHighlightParams) ([]lsp.DocumentHighlight, error) { documentManager := service.MustGet[workspace.DocumentManager](s.sc) uri := core.ParseURI(string(params.TextDocument.URI)) diff --git a/server/document_symbol_provider.go b/server/document_symbol_provider.go index eb5ed0c8..4b998eef 100644 --- a/server/document_symbol_provider.go +++ b/server/document_symbol_provider.go @@ -39,6 +39,14 @@ type DefaultDocumentSymbolProvider struct { filter DocumentSymbolFilter } +var _ DocumentSymbolProvider = (*DefaultDocumentSymbolProvider)(nil) +var _ DocumentStateRequirements = (*DefaultDocumentSymbolProvider)(nil) + +// RequiredState indicates that the default document symbol provider requires the document to be parsed. +func (p *DefaultDocumentSymbolProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateParsed, false +} + // NewDefaultDocumentSymbolProvider creates a provider using services from the container. func NewDefaultDocumentSymbolProvider(sc *service.Container) DocumentSymbolProvider { return &DefaultDocumentSymbolProvider{ diff --git a/server/folding_range_provider.go b/server/folding_range_provider.go index a88628a5..3fa56f89 100644 --- a/server/folding_range_provider.go +++ b/server/folding_range_provider.go @@ -66,6 +66,15 @@ type DefaultFoldingRangeProvider struct { filter FoldingRangeFilter } +// Ensure DefaultFoldingRangeProvider implements the expected interfaces. +var _ FoldingRangeProvider = (*DefaultFoldingRangeProvider)(nil) +var _ DocumentStateRequirements = (*DefaultFoldingRangeProvider)(nil) + +// RequiredState indicates that the default folding range provider requires the document to be parsed. +func (p *DefaultFoldingRangeProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateParsed, false +} + func NewDefaultFoldingRangeProvider(sc *service.Container) FoldingRangeProvider { return &DefaultFoldingRangeProvider{ sc: sc, diff --git a/server/hover_provider.go b/server/hover_provider.go index ba8f23d9..4396f2df 100644 --- a/server/hover_provider.go +++ b/server/hover_provider.go @@ -23,6 +23,15 @@ type DefaultHoverProvider struct { sc *service.Container } +// Ensure DefaultHoverProvider implements the expected interfaces. +var _ HoverProvider = (*DefaultHoverProvider)(nil) +var _ DocumentStateRequirements = (*DefaultHoverProvider)(nil) + +// RequiredState indicates that the default hover provider requires the document to be linked. +func (s *DefaultHoverProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateLinked, false +} + func NewDefaultHoverProvider(sc *service.Container) HoverProvider { return &DefaultHoverProvider{sc: sc} } diff --git a/server/references_finder.go b/server/references_finder.go index 7a4090c6..bcc2dc0a 100644 --- a/server/references_finder.go +++ b/server/references_finder.go @@ -33,6 +33,9 @@ type DefaultReferencesFinder struct { sc *service.Container } +// Ensure DefaultReferencesFinder implements the expected interfaces. +var _ ReferencesFinder = (*DefaultReferencesFinder)(nil) + func NewDefaultReferencesFinder(sc *service.Container) ReferencesFinder { return &DefaultReferencesFinder{sc: sc} } diff --git a/server/references_provider.go b/server/references_provider.go index 988d5d42..28eb720a 100644 --- a/server/references_provider.go +++ b/server/references_provider.go @@ -23,6 +23,16 @@ type DefaultReferencesProvider struct { sc *service.Container } +// Ensure DefaultReferencesProvider implements the expected interfaces. +var _ ReferencesProvider = (*DefaultReferencesProvider)(nil) +var _ DocumentStateRequirements = (*DefaultReferencesProvider)(nil) + +// RequiredState indicates that the default references provider requires the workspace +// to have reference information collected. +func (rf *DefaultReferencesProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateReferences, true +} + func NewDefaultReferencesProvider(sc *service.Container) ReferencesProvider { return &DefaultReferencesProvider{sc: sc} } diff --git a/server/rename_provider.go b/server/rename_provider.go index 5703b459..1434fba8 100644 --- a/server/rename_provider.go +++ b/server/rename_provider.go @@ -22,6 +22,16 @@ type DefaultRenameProvider struct { sc *service.Container } +// Ensure DefaultRenameProvider implements the expected interfaces. +var _ RenameProvider = (*DefaultRenameProvider)(nil) +var _ DocumentStateRequirements = (*DefaultRenameProvider)(nil) + +// RequiredState indicates that the default rename provider requires the workspace +// to have reference information collected. +func (rp *DefaultRenameProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateReferences, true +} + func NewDefaultRenameProvider(sc *service.Container) RenameProvider { return &DefaultRenameProvider{sc: sc} } diff --git a/server/server.go b/server/server.go index b91b8881..4f2cde09 100644 --- a/server/server.go +++ b/server/server.go @@ -9,6 +9,7 @@ import ( "log" "golang.org/x/exp/jsonrpc2" + core "typefox.dev/fastbelt" "typefox.dev/fastbelt/util/service" "typefox.dev/fastbelt/workspace" "typefox.dev/lsp" @@ -24,6 +25,18 @@ type InitializeParticipant interface { OnServerInitialize(params *lsp.ParamInitialize) } +// DocumentStateRequirements is an interface for services that require certain a document state +// to be reached before they can be used. If a service does not implement this interface, +// the language server will assume that the service can only run after the workspace has been +// fully built. +type DocumentStateRequirements interface { + // RequiredState returns the document state that must be reached before + // the service can be used. The boolean value indicates whether the full + // workspace needs to be in this state (true) or only the document being + // processed (false). + RequiredState() (core.DocumentState, bool) +} + // DefaultLanguageServer implements the [lsp.Server] interface type DefaultLanguageServer struct { sc *service.Container @@ -147,6 +160,19 @@ func (s *DefaultLanguageServer) DidSave(ctx context.Context, params *lsp.DidSave return nil } +func read(ctx context.Context, lock workspace.Lock, uri core.URI, service any, do func(ctx context.Context)) error { + if req, ok := service.(DocumentStateRequirements); ok { + var uris []core.URI + state, full := req.RequiredState() + if !full && uri != nil { + uris = append(uris, uri) + } + return lock.ReadAt(ctx, state, uris, do) + } else { + return lock.Read(ctx, do) + } +} + func (s *DefaultLanguageServer) Completion(ctx context.Context, params *lsp.CompletionParams) (*lsp.CompletionList, error) { var result *lsp.CompletionList var providerErr error @@ -158,7 +184,8 @@ func (s *DefaultLanguageServer) Completion(ctx context.Context, params *lsp.Comp if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, completion, func(ctx context.Context) { result, providerErr = completion.HandleCompletionRequest(ctx, params) }); err != nil { return nil, err @@ -177,7 +204,8 @@ func (s *DefaultLanguageServer) Definition(ctx context.Context, params *lsp.Defi if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, definition, func(ctx context.Context) { result, providerErr = definition.HandleDefinitionRequest(ctx, params) }); err != nil { return nil, err @@ -196,7 +224,8 @@ func (s *DefaultLanguageServer) References(ctx context.Context, params *lsp.Refe if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, references, func(ctx context.Context) { result, providerErr = references.HandleReferencesRequest(ctx, params) }); err != nil { return nil, err @@ -274,7 +303,8 @@ func (s *DefaultLanguageServer) DocumentHighlight(ctx context.Context, params *l if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, provider, func(ctx context.Context) { result, providerErr = provider.HandleDocumentHighlightRequest(ctx, params) }); err != nil { return nil, err @@ -295,7 +325,8 @@ func (s *DefaultLanguageServer) DocumentSymbol(ctx context.Context, params *lsp. if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, provider, func(ctx context.Context) { result, providerErr = provider.HandleDocumentSymbolRequest(ctx, params) }); err != nil { return nil, err @@ -313,7 +344,8 @@ func (s *DefaultLanguageServer) FoldingRange(ctx context.Context, params *lsp.Fo if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, provider, func(ctx context.Context) { result, providerErr = provider.HandleFoldingRangeRequest(ctx, params) }); err != nil { return nil, err @@ -334,7 +366,8 @@ func (s *DefaultLanguageServer) Hover(ctx context.Context, params *lsp.HoverPara if err != nil { return nil, err } - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, provider, func(ctx context.Context) { result, providerErr = provider.HandleHoverRequest(ctx, params) }); err != nil { return nil, err @@ -391,7 +424,7 @@ func (s *DefaultLanguageServer) Symbol(ctx context.Context, params *lsp.Workspac if err != nil { return nil, nil // No provider registered, return empty } - if err := lock.Read(ctx, func(ctx context.Context) { + if err := read(ctx, lock, nil, provider, func(ctx context.Context) { result, providerErr = provider.HandleWorkspaceSymbolRequest(ctx, params) }); err != nil { return nil, err @@ -433,7 +466,8 @@ func (s *DefaultLanguageServer) PrepareRename(ctx context.Context, params *lsp.P } var result *lsp.PrepareRenameResult var providerErr error - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, renameProvider, func(ctx context.Context) { result, providerErr = renameProvider.PrepareRenameRequest(ctx, params) }); err != nil { return nil, err @@ -460,7 +494,8 @@ func (s *DefaultLanguageServer) Rename(ctx context.Context, params *lsp.RenamePa } var result *lsp.WorkspaceEdit var providerErr error - if err := lock.Read(ctx, func(ctx context.Context) { + uri := core.ParseURI(string(params.TextDocument.URI)) + if err := read(ctx, lock, uri, renameProvider, func(ctx context.Context) { result, providerErr = renameProvider.HandleRenameRequest(ctx, params) }); err != nil { return nil, err diff --git a/server/workspace_symbol_provider.go b/server/workspace_symbol_provider.go index 41c9be19..8c31df9b 100644 --- a/server/workspace_symbol_provider.go +++ b/server/workspace_symbol_provider.go @@ -42,12 +42,21 @@ func (f *DefaultWorkspaceSymbolFilter) MaxSymbolCount() int { return 1000 } +// Ensure DefaultWorkspaceSymbolProvider implements the expected interfaces. +var _ WorkspaceSymbolProvider = (*DefaultWorkspaceSymbolProvider)(nil) +var _ DocumentStateRequirements = (*DefaultWorkspaceSymbolProvider)(nil) + // DefaultWorkspaceSymbolProvider implements WorkspaceSymbolProvider. type DefaultWorkspaceSymbolProvider struct { sc *service.Container filter WorkspaceSymbolFilter } +// RequiredState indicates that the default workspace symbol provider requires the workspace to be parsed. +func (f *DefaultWorkspaceSymbolProvider) RequiredState() (core.DocumentState, bool) { + return core.DocStateParsed, true +} + // NewDefaultWorkspaceSymbolProvider creates a provider using services from the container. func NewDefaultWorkspaceSymbolProvider(sc *service.Container) WorkspaceSymbolProvider { return &DefaultWorkspaceSymbolProvider{ diff --git a/test/doc_fixture.go b/test/doc_fixture.go index 37fd993d..d535a969 100644 --- a/test/doc_fixture.go +++ b/test/doc_fixture.go @@ -193,8 +193,8 @@ func (d *Doc) AssertNoDiagnostics() *Doc { // AssertState fails the test unless the document state includes the given flag. func (d *Doc) AssertState(flag core.DocumentState) *Doc { d.fixture.t.Helper() - if !d.Document.State.Has(flag) { - d.fixture.t.Errorf("fbtest: document state does not include %v (actual: %v)", flag, d.Document.State) + if !d.Document.State().Has(flag) { + d.fixture.t.Errorf("fbtest: document state does not include %v (actual: %v)", flag, d.Document.State()) } return d } diff --git a/test/doc_fixture_lsp.go b/test/doc_fixture_lsp.go index a4de5511..2c1d2eb3 100644 --- a/test/doc_fixture_lsp.go +++ b/test/doc_fixture_lsp.go @@ -59,7 +59,7 @@ func (d *Doc) RunRename(label string, newName string) *Doc { // Fully reset each document builder.Reset(doc, 0) } - if err := builder.Build(d.fixture.ctx, toUpdate, nil); err != nil { + if err := builder.Build(d.fixture.ctx, toUpdate); err != nil { d.fixture.t.Fatalf("fbtest: build failed after rename: %v", err) } return d diff --git a/test/fixture.go b/test/fixture.go index a1f8365b..7ee0cef2 100644 --- a/test/fixture.go +++ b/test/fixture.go @@ -144,7 +144,7 @@ func (f *Fixture) ParseURI(content, uri string) *Doc { documents := service.MustGet[workspace.DocumentManager](f.sc) documents.Set(doc) builder := service.MustGet[workspace.Builder](f.sc) - if err := builder.Build(f.ctx, []*core.Document{doc}, nil); err != nil { + if err := builder.Build(f.ctx, []*core.Document{doc}); err != nil { f.t.Fatalf("fbtest: build failed: %v", err) } return f.newDoc(doc, ranges, indices) @@ -177,7 +177,7 @@ func (f *Fixture) ParseAll(uriContentPairs ...string) []*Doc { results = append(results, f.newDoc(doc, ranges, indices)) } builder := service.MustGet[workspace.Builder](f.sc) - if err := builder.Build(f.ctx, coreDocs, nil); err != nil { + if err := builder.Build(f.ctx, coreDocs); err != nil { f.t.Fatalf("fbtest: build failed: %v", err) } return results diff --git a/workspace/builder.go b/workspace/builder.go index 852d2645..f2a3025e 100644 --- a/workspace/builder.go +++ b/workspace/builder.go @@ -20,14 +20,20 @@ import ( // linking, reference indexing, and validation. type Builder interface { // Build processes the provided documents through all build phases (parse, compute - // symbol table, link, validate). It should regularly check ctx for cancellation - // between phases. downgrade must be called by the implementation once phases 1 and 2 - // (the write phase) are complete; this transitions the workspace lock to readable so - // that read requests can proceed while phase 3 (validation) runs. - Build(ctx context.Context, docs []*core.Document, downgrade func()) error + // symbol table, link, validate). It must be called under [Lock.Write] and should + // regularly check ctx for cancellation between phases. Calling Build marks the + // transition from the write's mutation phase to its build phase (see [Lock.Write]): + // implementations signal [Lock.StateChanged] as document states advance so that + // [Lock.ReadAt] requests are admitted as soon as the states they need are reached, + // and report the workspace-wide floor after each phase barrier. The floor relies + // on every document outside docs already being complete, which the caller must + // ensure (the updater collects every incomplete document into the build set). + Build(ctx context.Context, docs []*core.Document) error // Reset selectively clears build results of a document. The state parameter is a // bitmask of states to keep; for every bit that is not set, the corresponding document // fields are reset to their initial values and the bit is cleared from doc.State. + // Reset must be called under [Lock.Write], during the mutation phase - that is, + // before Build starts advancing document states. Reset(doc *core.Document, state core.DocumentState) // AddBuildStepListener registers a listener to be called after documents complete the // specified build steps. The states parameter is a bitmask, so multiple steps can be @@ -60,7 +66,19 @@ func NewDefaultBuilder(sc *service.Container) Builder { return &DefaultBuilder{sc: sc} } -func (s *DefaultBuilder) Build(ctx context.Context, docs []*core.Document, downgrade func()) error { +func (s *DefaultBuilder) Build(ctx context.Context, docs []*core.Document) error { + lock := service.MustGet[Lock](s.sc) + // Build has started, signal the lock to be ready to admit readers after + // advancing document states. + lock.StateChanged(0) + // Wakes up any ReadAt calls to make sure they can see the new document state. + // Documents might reach the requested state before the full workspace does. + advance := func(doc *core.Document, state core.DocumentState) { + doc.SetState(doc.State().With(state)) + lock.StateChanged(0) + s.notifyListeners(ctx, state, doc) + } + // PHASE 1: Parse, and compute exports (parallel per document). parser := service.MustGet[DocumentParser](s.sc) exporter := service.MustGet[linking.SymbolExporter](s.sc) @@ -69,25 +87,28 @@ func (s *DefaultBuilder) Build(ctx context.Context, docs []*core.Document, downg return } // STEP 1.1: Parse the document and create the AST. - if !doc.State.Has(core.DocStateParsed) { + if !doc.State().Has(core.DocStateParsed) { parser.Parse(doc) - doc.State = doc.State.With(core.DocStateParsed) - s.notifyListeners(ctx, core.DocStateParsed, doc) + advance(doc, core.DocStateParsed) } if ctx.Err() != nil { return } // STEP 1.2: Compute the exported symbols for cross-document references. - if !doc.State.Has(core.DocStateExportedSymbols) { + if !doc.State().Has(core.DocStateExportedSymbols) { exporter.ExportSymbols(ctx, doc) - doc.State = doc.State.With(core.DocStateExportedSymbols) - s.notifyListeners(ctx, core.DocStateExportedSymbols, doc) + advance(doc, core.DocStateExportedSymbols) } }) if err := ctx.Err(); err != nil { return err } + // Phase 1 barrier: every document in docs is now parsed with exports + // computed, and documents outside docs were already complete when the + // updater collected the build set. Report the workspace floor so + // workspace-wide ReadAt calls are admitted. + lock.StateChanged(core.DocStateParsed | core.DocStateExportedSymbols) // PHASE 2: Compute imported/local symbols and link (parallel per document). // This requires the exported symbols of all documents to be available. @@ -101,55 +122,43 @@ func (s *DefaultBuilder) Build(ctx context.Context, docs []*core.Document, downg return } // STEP 2.1: Collect imported symbols from all other documents. - if !doc.State.Has(core.DocStateImportedSymbols) { + if !doc.State().Has(core.DocStateImportedSymbols) { allDocs := documentManager.All() importer.ImportSymbols(ctx, doc, allDocs) - doc.State = doc.State.With(core.DocStateImportedSymbols) - s.notifyListeners(ctx, core.DocStateImportedSymbols, doc) + advance(doc, core.DocStateImportedSymbols) } if ctx.Err() != nil { return } // STEP 2.2: Compute the local symbols for intra-document references. - if !doc.State.Has(core.DocStateLocalSymbols) { + if !doc.State().Has(core.DocStateLocalSymbols) { localSymbols.LocalSymbols(ctx, doc) - doc.State = doc.State.With(core.DocStateLocalSymbols) - s.notifyListeners(ctx, core.DocStateLocalSymbols, doc) + advance(doc, core.DocStateLocalSymbols) } if ctx.Err() != nil { return } // STEP 2.3: Link the document to resolve all references. - if !doc.State.Has(core.DocStateLinked) { + if !doc.State().Has(core.DocStateLinked) { linker.Link(ctx, doc) - doc.State = doc.State.With(core.DocStateLinked) - s.notifyListeners(ctx, core.DocStateLinked, doc) + advance(doc, core.DocStateLinked) } if ctx.Err() != nil { return } // STEP 2.4: Provide reference descriptions for the document. - if !doc.State.Has(core.DocStateReferences) { + if !doc.State().Has(core.DocStateReferences) { referenceDescriptions.ReferenceDescriptions(ctx, doc) - doc.State = doc.State.With(core.DocStateReferences) - s.notifyListeners(ctx, core.DocStateReferences, doc) + advance(doc, core.DocStateReferences) } }) if err := ctx.Err(); err != nil { - // Important note: Do not downgrade the lock here! - // If we downgrade the lock here, we would allow read access to - // the workspace while the documents are in an inconsistent state. - // In most cases, the error has been triggered by a new change, - // which will trigger a new build with a re-acquired read-lock. return err } - - // Transition from write phase to readable: releases the exclusive lock so - // read requests can proceed while validation (phase 3) runs concurrently. - if downgrade != nil { - downgrade() - } + // Phase 2 barrier: the whole workspace is now linked and indexed. + lock.StateChanged(core.DocStateImportedSymbols | core.DocStateLocalSymbols | + core.DocStateLinked | core.DocStateReferences) // PHASE 3: Run custom validations (parallel per document). validator := service.MustGet[DocumentValidator](s.sc) @@ -157,18 +166,22 @@ func (s *DefaultBuilder) Build(ctx context.Context, docs []*core.Document, downg if ctx.Err() != nil { return } - if !doc.State.Has(core.DocStateValidated) { + if !doc.State().Has(core.DocStateValidated) { diagnostics := validator.Validate(ctx, doc, "on-save") if ctx.Err() != nil { return } doc.Diagnostics = diagnostics - doc.State = doc.State.With(core.DocStateValidated) - s.notifyListeners(ctx, core.DocStateValidated, doc) + advance(doc, core.DocStateValidated) } }) - return ctx.Err() + if err := ctx.Err(); err != nil { + return err + } + // Phase 3 barrier: the whole workspace is validated. + lock.StateChanged(core.DocStateValidated) + return nil } func (s *DefaultBuilder) Reset(doc *core.Document, state core.DocumentState) { @@ -204,7 +217,7 @@ func (s *DefaultBuilder) Reset(doc *core.Document, state core.DocumentState) { case !state.Has(core.DocStateValidated): doc.Diagnostics = []*core.Diagnostic{} } - doc.State = doc.State & state + doc.SetState(doc.State() & state) } func (s *DefaultBuilder) AddBuildStepListener(states core.DocumentState, listener BuildStepListener) { diff --git a/workspace/doc.go b/workspace/doc.go index d057f674..8d9b30ac 100644 --- a/workspace/doc.go +++ b/workspace/doc.go @@ -23,7 +23,7 @@ // registers the framework defaults: // // - [Initializer] — loads files matching [FileExtensions] from workspace folders on startup -// - [Lock] — read/write coordination with atomic write-to-read downgrade +// - [Lock] — read/write coordination with per-document state admission for readers // - [Builder] — runs the build pipeline that takes documents to a linked, validated state // - [DocumentManager] — concurrent in-memory store of all documents, keyed by URI // - [DocumentUpdater] — entry point for edits; serializes mutations and triggers builds @@ -52,11 +52,12 @@ // [typefox.dev/fastbelt/linking], which the builder resolves from the container; // see that package for how each step works. // -// Phases 1 and 2 are the write phase: they mutate document data and require -// exclusive access. Phase 3 only reads document data, so between phase 2 and -// phase 3 the builder calls the downgrade function passed to [Builder.Build], -// releasing exclusive access (see Concurrency below) so that read requests can -// proceed while validation runs. +// The whole build runs under the exclusive [Lock.Write], but requests do not +// have to wait for it to finish: each completed step is published through the +// document's [typefox.dev/fastbelt.DocumentState], and [Lock.ReadAt] admits +// readers as soon as the documents they target reach the states they need +// (see Concurrency below). A document-symbol request, for example, only needs +// a parsed AST and is served while linking and validation are still running. // // Because each step is guarded by its [typefox.dev/fastbelt.DocumentState] bit, // builds are incremental. [Builder.Reset] clears selected steps of a document @@ -82,17 +83,31 @@ // the context of an in-progress build, so superseded builds stop quickly while // the latest one runs to completion. // -// Read-only requests such as completion, hover, and go-to-definition run under -// [Lock.Read] and consult [DocumentManager] for the current document state. +// Read-only requests run under [Lock.Read] when they may touch arbitrary +// documents (find references, workspace symbols), or under [Lock.ReadAt] when +// they target a known set of documents and a known build state. // // # Concurrency // -// [Lock] serializes writes against reads and provides the atomic write-to-read -// downgrade the build pipeline relies on. [Lock.Write] grants exclusive access -// for the write phase, then the downgrade callback atomically converts that -// exclusive hold into a shared read hold, with no window for another writer to -// intervene. This guarantees validation (phase 3) observes a consistent, -// fully linked snapshot and completes before the next write phase begins. -// Writes have priority: a pending write blocks new readers, and starting a -// write cancels any write still in progress so the freshest edit wins. +// [Lock.Write] grants exclusive access for a build cycle. A write starts in +// the mutation phase, in which documents may be created, deleted, and reset; +// once [Builder.Build] begins, document states only advance, and every +// advance is signalled through [Lock.StateChanged]. [Lock.ReadAt] relies on +// this monotonicity: it admits a reader as soon as every requested document +// carries the requested state bits, even while the build is still writing +// later phases or other documents. Setting a state bit (an atomic store) +// publishes all data written by that step, so an admitted reader always +// observes complete data for the states it requested — provided it touches +// only the documents it listed and only the data those states produce. +// For whole-workspace access at a specific state, ReadAt accepts an empty +// document list: it is then admitted based on the workspace floor — the state +// every document is guaranteed to have reached — which the builder reports +// through [Lock.StateChanged] at each phase barrier, without any document +// scan. [Lock.Read] is simply the state-agnostic variant: it waits until no +// write is active or pending. +// +// Writes have priority: a pending write blocks new readers (including +// ReadAt), and starting a write cancels any write still in progress so the +// freshest edit wins. A write acquires the lock only after all readers have +// drained, so a reader admitted via ReadAt never observes a document reset. package workspace diff --git a/workspace/document_updater.go b/workspace/document_updater.go index 3e93c51a..7b6042f6 100644 --- a/workspace/document_updater.go +++ b/workspace/document_updater.go @@ -48,7 +48,7 @@ func (s *DefaultDocumentUpdater) Update(ctx context.Context, changed []textdoc.H // Write cancels any previous pending or in-progress build and issues a // fresh context. The outer ctx is from jsonrpc2 and has a different // lifetime than the build. - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { + lock.Write(context.Background(), func(ctx context.Context) { changedURIs := make(collections.Set[string], len(changed)+len(deleted)) for _, handle := range changed { doc := core.NewDocument(handle) @@ -74,12 +74,12 @@ func (s *DefaultDocumentUpdater) Update(ctx context.Context, changed []textdoc.H if !changedURIs.Has(doc.URI.StringUnencoded()) && changeImpact.Affected(doc, changedURIs) { builder.Reset(doc, keepState) } - if !doc.State.IsComplete() { + if !doc.State().IsComplete() { docs = append(docs, doc) } } - if err := builder.Build(ctx, docs, downgrade); err != nil { + if err := builder.Build(ctx, docs); err != nil { if ctx.Err() == nil { log.Printf("build failed: %v", err) } diff --git a/workspace/lock.go b/workspace/lock.go index 0b0ee836..b0fced5f 100644 --- a/workspace/lock.go +++ b/workspace/lock.go @@ -7,58 +7,85 @@ package workspace import ( "context" "sync" + + core "typefox.dev/fastbelt" + "typefox.dev/fastbelt/util/service" ) // Lock controls read/write access to workspace [core.Document] data. -// LSP request handlers use [Lock.Read]; document updates and builds use -// [Lock.Write] with an atomic downgrade to shared access for validation. +// +// Builds run under the exclusive [Lock.Write]. LSP request handlers use +// [Lock.Read] for whole-workspace access, or [Lock.ReadAt] to run against a +// known set of documents as soon as those documents reach the required build +// states — potentially while a build is still writing other documents. type Lock interface { - // Write cancels any pending or in-progress write, then acquires an exclusive - // lock and calls do with a fresh context and a downgrade function. Calling - // downgrade atomically transitions from the exclusive write lock to a shared - // read lock, allowing reads to proceed while the caller's read phase continues. - // downgrade is idempotent; a safety net ensures it is always called. - Write(ctx context.Context, do func(ctx context.Context, downgrade func())) - // Read acquires a shared lock, calls do, then releases the lock. - // It blocks while a write is in progress or pending. - // Read returns ctx.Err() if ctx is cancelled while waiting to acquire the lock. + // Write cancels any pending or in-progress write, then acquires an + // exclusive lock and calls do with a fresh context. do is always called, + // even if the context was cancelled by a newer write, so that document + // mutations are never silently dropped. Cancelled write actions should + // skip expensive work by checking ctx.Err(). + Write(ctx context.Context, do func(ctx context.Context)) + // Read acquires a shared lock on the whole workspace, calls do, then + // releases the lock. It blocks while a write is in progress or pending. + // Read returns ctx.Err() if ctx is cancelled while waiting. Read(ctx context.Context, do func(ctx context.Context)) error + // ReadAt acquires a shared lock scoped to the given documents. It is + // admitted as soon as every URI resolves to a document whose state covers + // states, even while a build is writing. A URI with no document yet counts + // as not ready and waits for the build that creates it. ReadAt returns + // ctx.Err() if ctx is cancelled while waiting. + // + // If uris is empty, ReadAt is scoped to the whole workspace: it is + // admitted once every document in the workspace has reached states, as + // reported to [Lock.StateChanged] by the build. + ReadAt(ctx context.Context, states core.DocumentState, uris []core.URI, do func(ctx context.Context)) error + // StateChanged signals that document build states advanced, waking + // pending ReadAt calls so they re-check their documents. current is the + // build state that every document in the workspace is guaranteed to have + // reached (the workspace floor), or 0 if no workspace-wide guarantee can + // be made; the floor accumulates monotonically until the next write + // acquires the lock and admits workspace-wide ReadAt calls. + // + // [Builder.Build] calls it once when the build phase begins, after every + // per-document state advance, and with the reached floor after each phase + // barrier. Other Lock users normally never call it. + StateChanged(current core.DocumentState) } // DefaultLock is the default implementation of [Lock]. // -// The key property is atomic write-to-read downgrade: when downgrade is called, -// the caller atomically transitions from holding the exclusive write lock to holding -// a shared read lock, with no window in which a new writer could sneak in. This -// guarantees that phase 3 (validation) completes under a read lock before any -// subsequent write phase can begin. -// -// Cancellation only cancels the context passed to do - every write that enters -// Write always calls do, even if its context was cancelled by a newer write. -// This ensures that document mutations inside do (e.g. applying text changes) -// are never silently dropped; the cancelled do simply skips expensive work by -// checking ctx.Err() before each phase. +// Writes have priority: a pending write blocks new readers (including ReadAt), +// and starting a write cancels any write still in progress so the freshest +// edit wins. The next write acquires the lock only after all readers — shared +// and state-scoped — have drained, so readers never observe document resets. type DefaultLock struct { - mu sync.Mutex - cond *sync.Cond - writeHeld bool // exclusive write phase is active - writeWaiters int // number of goroutines waiting to acquire the write lock - readers int // number of active shared read lock holders - readyCh chan struct{} // closed when !writeHeld && writeWaiters==0; replaced each cycle - cancelWrite context.CancelFunc // cancels the current pending or in-progress write + sc *service.Container + docs DocumentManager // lazily resolved from sc on first ReadAt + + mu sync.Mutex + cond *sync.Cond + writeHeld bool // exclusive write phase is active + writeWaiters int // number of goroutines waiting to acquire the write lock + readers int // number of active shared lock holders (Read and ReadAt) + stateAdvanced bool // current write entered the build phase; document states only advance + workspaceState core.DocumentState // floor reached by every document; reset when a write acquires the lock + readyCh chan struct{} // closed when !writeHeld && writeWaiters==0; replaced each cycle + stateCh chan struct{} // closed and replaced whenever ReadAt admission may have changed + cancelWrite context.CancelFunc // cancels the current pending or in-progress write } -// NewDefaultLock returns a [Lock] with write-priority scheduling and atomic -// write-to-read downgrade. -func NewDefaultLock() Lock { - l := &DefaultLock{} +// NewDefaultLock returns a [Lock] with write-priority scheduling and +// per-document state admission for ReadAt. +func NewDefaultLock(sc *service.Container) Lock { + l := &DefaultLock{sc: sc} l.cond = sync.NewCond(&l.mu) l.readyCh = make(chan struct{}) close(l.readyCh) // initially readable + l.stateCh = make(chan struct{}) return l } -func (l *DefaultLock) Write(ctx context.Context, do func(ctx context.Context, downgrade func())) { +func (l *DefaultLock) Write(ctx context.Context, do func(ctx context.Context)) { ctx, cancel := context.WithCancel(ctx) l.mu.Lock() @@ -80,38 +107,25 @@ func (l *DefaultLock) Write(ctx context.Context, do func(ctx context.Context, do } l.writeWaiters-- l.writeHeld = true + // Back in the mutation phase: ReadAt must not trust document states, and + // the workspace floor no longer holds (documents may be reset or created). + l.stateAdvanced = false + l.workspaceState = 0 l.mu.Unlock() - var once sync.Once - downgrade := func() { - once.Do(func() { - l.mu.Lock() - defer l.mu.Unlock() - l.writeHeld = false - // Downgrade: atomically acquire a read lock before releasing the write - // lock. This leaves no window in which a new writer could start before - // the caller's read phase (phase 3 / validation) has completed. - l.readers++ - if l.writeWaiters == 0 { - // No writer is waiting - unblock pending reads. - close(l.readyCh) - } - l.cond.Broadcast() // wake any writer waiting in cond.Wait - }) - } - defer func() { - downgrade() // safety net: ensures the write lock is always released l.mu.Lock() - // Release the read lock that downgrade acquired. - l.readers-- - if l.readers == 0 { - l.cond.Broadcast() // wake any writer waiting for readers to drain + l.writeHeld = false + if l.writeWaiters == 0 { + // No writer is waiting - unblock pending reads. + close(l.readyCh) } + l.wakeReadAtLocked() + l.cond.Broadcast() // wake any writer waiting in cond.Wait l.mu.Unlock() }() - do(ctx, downgrade) + do(ctx) } func (l *DefaultLock) Read(ctx context.Context, do func(ctx context.Context)) error { @@ -132,15 +146,82 @@ func (l *DefaultLock) Read(ctx context.Context, do func(ctx context.Context)) er } } - defer func() { + defer l.releaseReader() + + do(ctx) + return nil +} + +func (l *DefaultLock) ReadAt(ctx context.Context, states core.DocumentState, uris []core.URI, do func(ctx context.Context)) error { + for { l.mu.Lock() - l.readers-- - if l.readers == 0 { - l.cond.Broadcast() + // Currently, no write is pending + // A pending write has priority over readers + if l.writeWaiters == 0 && + // The current write is ready to admit readers + (!l.writeHeld || l.stateAdvanced) && + // Every requested document is at the requested states + l.docsReadyLocked(states, uris) { + l.readers++ + l.mu.Unlock() + break } + ch := l.stateCh l.mu.Unlock() - }() + + select { + case <-ch: // states advanced or a write cycle ended; re-check + case <-ctx.Done(): + return ctx.Err() + } + } + + defer l.releaseReader() do(ctx) return nil } + +func (l *DefaultLock) StateChanged(current core.DocumentState) { + l.mu.Lock() + l.stateAdvanced = true + l.workspaceState = l.workspaceState.With(current) + l.wakeReadAtLocked() + l.mu.Unlock() +} + +// wakeReadAtLocked wakes all ReadAt calls waiting for admission so they +// re-check their documents. Callers must hold l.mu. +func (l *DefaultLock) wakeReadAtLocked() { + close(l.stateCh) + l.stateCh = make(chan struct{}) +} + +// docsReadyLocked reports whether every URI resolves to a document whose state +// covers states. An empty uris list means the whole workspace and is checked +// against the floor reported through StateChanged, without scanning documents. +// Callers must hold l.mu. +func (l *DefaultLock) docsReadyLocked(states core.DocumentState, uris []core.URI) bool { + if len(uris) == 0 { + return l.workspaceState.Has(states) + } + if l.docs == nil { + l.docs = service.MustGet[DocumentManager](l.sc) + } + for _, uri := range uris { + doc := l.docs.Get(uri) + if doc == nil || !doc.State().Has(states) { + return false + } + } + return true +} + +func (l *DefaultLock) releaseReader() { + l.mu.Lock() + l.readers-- + if l.readers == 0 { + l.cond.Broadcast() // wake any writer waiting for readers to drain + } + l.mu.Unlock() +} diff --git a/workspace/lock_test.go b/workspace/lock_test.go index 1afd1999..95e9acfc 100644 --- a/workspace/lock_test.go +++ b/workspace/lock_test.go @@ -11,487 +11,625 @@ import ( "time" "github.com/stretchr/testify/assert" + core "typefox.dev/fastbelt" + "typefox.dev/fastbelt/util/service" ) const shortWait = 20 * time.Millisecond const longWait = 2 * time.Second +// signal is a one-shot notification between the test goroutine and lock +// operations. fire is idempotent. +type signal struct { + ch chan struct{} + once sync.Once +} + +func newSignal() *signal { return &signal{ch: make(chan struct{})} } + +func (s *signal) fire() { s.once.Do(func() { close(s.ch) }) } + +// await blocks until the signal fires, failing the test after longWait. +func (s *signal) await(t *testing.T, msg string) { + t.Helper() + select { + case <-s.ch: + case <-time.After(longWait): + t.Fatal(msg) + } +} + +// assertPending verifies the signal does not fire within shortWait. +func (s *signal) assertPending(t *testing.T, msg string) { + t.Helper() + select { + case <-s.ch: + t.Fatal(msg) + case <-time.After(shortWait): + } +} + +// lockHarness bundles a lock and its document manager with helpers to run +// lock operations in goroutines under explicit admission and release control. +type lockHarness struct { + t *testing.T + lock Lock + dm DocumentManager +} + +func newLockHarness(t *testing.T) *lockHarness { + sc := service.NewContainer() + dm := NewDefaultDocumentManager(sc) + service.Put(sc, dm) + sc.Seal() + return &lockHarness{t: t, lock: NewDefaultLock(sc), dm: dm} +} + +// doc creates a document with the given state and registers it. +func (h *lockHarness) doc(uri string, state core.DocumentState) *core.Document { + h.t.Helper() + doc, err := core.NewDocumentFromString(uri, "test", "") + if err != nil { + h.t.Fatal(err) + } + doc.SetState(state) + h.dm.Set(doc) + return doc +} + +// lockOp is a lock operation (Write, Read, or ReadAt) running in its own +// goroutine. Its callback fires entered once the lock admits it, then holds +// the lock until release fires; done fires when the lock call has returned. +type lockOp struct { + t *testing.T + entered *signal + release *signal + done *signal + ctx context.Context // ctx passed to the callback; valid once entered fired + err error // result of Read/ReadAt; valid once done fired +} + +func (h *lockHarness) newOp() *lockOp { + return &lockOp{t: h.t, entered: newSignal(), release: newSignal(), done: newSignal()} +} + +// enter is the callback run under the lock: it records the callback context, +// reports admission, and holds the lock until the test releases it. +func (op *lockOp) enter(ctx context.Context) { + op.ctx = ctx + op.entered.fire() + <-op.release.ch +} + +// startWrite runs lock.Write in a goroutine and returns its handle. +func (h *lockHarness) startWrite(ctx context.Context) *lockOp { + op := h.newOp() + go func() { + h.lock.Write(ctx, op.enter) + op.done.fire() + }() + return op +} + +// startRead runs lock.Read in a goroutine and returns its handle. +func (h *lockHarness) startRead(ctx context.Context) *lockOp { + op := h.newOp() + go func() { + op.err = h.lock.Read(ctx, op.enter) + op.done.fire() + }() + return op +} + +// startReadAt runs lock.ReadAt in a goroutine and returns its handle. +func (h *lockHarness) startReadAt(ctx context.Context, states core.DocumentState, uris []core.URI) *lockOp { + op := h.newOp() + go func() { + op.err = h.lock.ReadAt(ctx, states, uris, op.enter) + op.done.fire() + }() + return op +} + +// awaitEntered waits until the lock admits the operation. +func (op *lockOp) awaitEntered(msg string) { + op.t.Helper() + op.entered.await(op.t, msg) +} + +// assertBlocked verifies the lock does not admit the operation within shortWait. +func (op *lockOp) assertBlocked(msg string) { + op.t.Helper() + op.entered.assertPending(op.t, msg) +} + +// assertNotEntered verifies (without waiting) that the callback never ran. +func (op *lockOp) assertNotEntered(msg string) { + op.t.Helper() + select { + case <-op.entered.ch: + op.t.Fatal(msg) + default: + } +} + +// awaitDone waits for the lock call to return; its result is left in op.err. +func (op *lockOp) awaitDone(msg string) { + op.t.Helper() + op.done.await(op.t, msg) +} + +// finish releases the callback, waits for the lock call to return, and +// asserts that it succeeded. Operations expected to fail use awaitDone and +// inspect op.err instead. +func (op *lockOp) finish() { + op.t.Helper() + op.release.fire() + op.awaitDone("lock operation did not finish after release") + assert.NoError(op.t, op.err) +} + // TestReadRunsDoAndReturnsNil verifies the basic happy path: Read calls do and returns nil. func TestReadRunsDoAndReturnsNil(t *testing.T) { - lock := NewDefaultLock() + h := newLockHarness(t) called := false - err := lock.Read(context.Background(), func(ctx context.Context) { called = true }) + err := h.lock.Read(context.Background(), func(ctx context.Context) { called = true }) assert.NoError(t, err) assert.True(t, called) } // TestConcurrentReads verifies that multiple Read calls can hold the lock simultaneously. func TestConcurrentReads(t *testing.T) { - lock := NewDefaultLock() + h := newLockHarness(t) const n = 10 - inside := make(chan struct{}, n) - release := make(chan struct{}) - - var wg sync.WaitGroup - for range n { - wg.Go(func() { - err := lock.Read(context.Background(), func(ctx context.Context) { - inside <- struct{}{} - // Wait for the test to signal release before exiting - // so all readers are inside simultaneously. - <-release - }) - assert.NoError(t, err) - }) + readers := make([]*lockOp, n) + for i := range readers { + readers[i] = h.startRead(context.Background()) } - - // All n readers must be inside simultaneously. - for range n { - select { - case <-inside: - case <-time.After(longWait): - t.Fatal("timed out waiting for concurrent readers") - } + // All n readers must be inside simultaneously: each is admitted while the + // others still hold their read lock. + for _, r := range readers { + r.awaitEntered("timed out waiting for concurrent readers") + } + for _, r := range readers { + r.finish() } - close(release) - wg.Wait() } -// TestReadContextCancelledBeforeWrite verifies that Read returns ctx.Err() when +// TestReadContextCancelledBeforeAcquire verifies that Read returns ctx.Err() when // the context is already cancelled before the read lock is acquired. func TestReadContextCancelledBeforeAcquire(t *testing.T) { - lock := NewDefaultLock() + h := newLockHarness(t) - // Hold write phase so the reader must wait. - inWrite := make(chan struct{}) - releaseWrite := make(chan struct{}) - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - close(inWrite) - <-releaseWrite - }) - }() - <-inWrite + // Hold the write lock so the reader must wait. + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") ctx, cancel := context.WithCancel(context.Background()) cancel() // already cancelled called := false - err := lock.Read(ctx, func(ctx context.Context) { called = true }) + err := h.lock.Read(ctx, func(ctx context.Context) { called = true }) assert.ErrorIs(t, err, context.Canceled) assert.False(t, called) - close(releaseWrite) + w.finish() } // TestReadContextCancelledWhileWaiting verifies that a blocked Read returns // ctx.Err() when its context is cancelled while waiting for a write to finish. func TestReadContextCancelledWhileWaiting(t *testing.T) { - lock := NewDefaultLock() + h := newLockHarness(t) - inWrite := make(chan struct{}) - releaseWrite := make(chan struct{}) - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - close(inWrite) - <-releaseWrite - }) - }() - <-inWrite + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") ctx, cancel := context.WithCancel(context.Background()) - readErr := make(chan error, 1) - called := false - go func() { - readErr <- lock.Read(ctx, func(ctx context.Context) { called = true }) - }() + r := h.startRead(ctx) + r.assertBlocked("read proceeded while write was active") - time.Sleep(shortWait) cancel() + r.awaitDone("Read did not return after context cancellation") + assert.ErrorIs(t, r.err, context.Canceled) + r.assertNotEntered("read callback ran despite cancellation") - select { - case err := <-readErr: - assert.ErrorIs(t, err, context.Canceled) - assert.False(t, called) - case <-time.After(longWait): - t.Fatal("Read did not return after context cancellation") - } + w.finish() +} + +// TestWriteBlocksReadsUntilDone verifies that reads are blocked while a write +// is active and unblocked once the write's do returns. +func TestWriteBlocksReadsUntilDone(t *testing.T) { + h := newLockHarness(t) - close(releaseWrite) + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") + + r := h.startRead(context.Background()) + r.assertBlocked("read proceeded while write was active") + + w.finish() + r.awaitEntered("read did not proceed after write finished") + r.finish() } -// TestWriteBlocksReadsUntilDowngrade verifies that reads are blocked during -// the write phase and unblocked once downgrade is called. -func TestWriteBlocksReadsUntilDowngrade(t *testing.T) { - lock := NewDefaultLock() +// TestWriteWaitsForActiveReaders verifies that Write only acquires the lock +// after all in-progress Read calls have completed. +func TestWriteWaitsForActiveReaders(t *testing.T) { + h := newLockHarness(t) - inWritePhase := make(chan struct{}) - doDowngrade := make(chan struct{}) - writeDoRunning := make(chan struct{}) + r := h.startRead(context.Background()) + r.awaitEntered("read was not admitted") - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - close(inWritePhase) - <-doDowngrade - downgrade() - // Hold the read lock so we can observe reads running concurrently. - <-writeDoRunning - }) - }() - <-inWritePhase + w := h.startWrite(context.Background()) + w.assertBlocked("write started before the reader released") - // A read started during the write phase must block. - readReached := make(chan struct{}) - go func() { - err := lock.Read(context.Background(), func(ctx context.Context) { close(readReached) }) - assert.NoError(t, err) - }() + r.finish() + w.awaitEntered("write never started after reader finished") + w.finish() +} - select { - case <-readReached: - t.Fatal("read proceeded before downgrade") - case <-time.After(shortWait): - // expected: blocked - } +// TestNewWriteCancelsPendingWrite verifies that when a second Write call arrives +// while the first is still waiting to acquire the lock, the first still runs do +// (so document mutations are never lost) but receives a cancelled context so it +// can skip expensive work such as building. +func TestNewWriteCancelsPendingWrite(t *testing.T) { + h := newLockHarness(t) + + // Hold a read lock so both writers must wait. + r := h.startRead(context.Background()) + r.awaitEntered("read was not admitted") + + w1 := h.startWrite(context.Background()) + time.Sleep(shortWait) // let W1 register with the lock before W2 arrives + w2 := h.startWrite(context.Background()) + + // Admission order between two queued writers is unspecified: pre-release + // both callbacks and assert only on the contexts they received. + w1.release.fire() + w2.release.fire() + r.finish() + w1.awaitDone("W1 did not complete") + w2.awaitDone("W2 did not complete") + + // W1 must have run (mutations must not be lost) but with a cancelled context. + assert.ErrorIs(t, w1.ctx.Err(), context.Canceled, "W1 should have received a cancelled context") + // W2 is the newest writer and must receive a live context. + assert.NoError(t, w2.ctx.Err(), "W2 should have received a fresh context") +} - // Signal downgrade - the read should now unblock. - close(doDowngrade) +// TestNewWriteCancelsActiveWrite verifies that a second Write call cancels the +// first while its do callback is actively running, and then runs itself. +func TestNewWriteCancelsActiveWrite(t *testing.T) { + h := newLockHarness(t) + w1 := h.startWrite(context.Background()) + w1.awaitEntered("W1 was not admitted") + + // Starting W2 cancels W1's context while W1 is still running. + w2 := h.startWrite(context.Background()) select { - case <-readReached: - // good: read proceeded after downgrade + case <-w1.ctx.Done(): case <-time.After(longWait): - t.Fatal("read did not proceed after downgrade") + t.Fatal("W1's context was not cancelled by the new write") } - close(writeDoRunning) + w2.assertBlocked("W2 started before W1 finished") + w1.finish() + w2.awaitEntered("W2 was never admitted after W1 finished") + w2.finish() } -// TestDowngradeAllowsReadsWhileDoStillRuns verifies that do continues running -// (e.g. validation / phase 3) after downgrade while readers proceed concurrently. -func TestDowngradeAllowsReadsWhileDoStillRuns(t *testing.T) { - lock := NewDefaultLock() +// TestWriteHasPriorityOverQueuedRead verifies the ordering: +// Write 1 enters -> Read queues up -> Write 2 arrives -> Write 2 runs before the Read. +func TestWriteHasPriorityOverQueuedRead(t *testing.T) { + h := newLockHarness(t) + + w1 := h.startWrite(context.Background()) + w1.awaitEntered("W1 was not admitted") - readProceedDone := make(chan struct{}) - doFinish := make(chan struct{}) + // Read: queues up while Write 1 holds the lock. + r := h.startRead(context.Background()) + r.assertBlocked("read proceeded while W1 was active") - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - downgrade() - // do is still running (phase 3) - wait for the concurrent reader. - <-readProceedDone - close(doFinish) - }) - }() + // Write 2: arrives after the Read is already queued. + w2 := h.startWrite(context.Background()) + time.Sleep(shortWait) // let W2 register as a waiter + + w1.finish() + w2.awaitEntered("W2 was never admitted") + r.assertNotEntered("the queued read overtook the pending write") + w2.finish() + r.awaitEntered("read did not proceed after W2 finished") + r.finish() +} + +// TestReadAtImmediateWhenDocumentReady verifies that ReadAt runs immediately +// when the workspace is idle and the document already has the requested states. +func TestReadAtImmediateWhenDocumentReady(t *testing.T) { + h := newLockHarness(t) + doc := h.doc("file:///a.test", core.DocStateParsed) - // The read should proceed immediately (downgrade was called synchronously above). - err := lock.Read(context.Background(), func(ctx context.Context) { - close(readProceedDone) + called := false + err := h.lock.ReadAt(context.Background(), core.DocStateParsed, []core.URI{doc.URI}, func(ctx context.Context) { + called = true }) assert.NoError(t, err) + assert.True(t, called) +} - select { - case <-doFinish: - case <-time.After(longWait): - t.Fatal("write do did not finish after concurrent read completed") - } +// TestReadAtAdmittedDuringBuildPhase is the core behavior: a ReadAt whose +// document has reached the requested state is admitted while a write is still +// running, whereas a plain Read stays blocked. +func TestReadAtAdmittedDuringBuildPhase(t *testing.T) { + h := newLockHarness(t) + doc := h.doc("file:///a.test", 0) + + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") + // Build phase: parse completes. + doc.SetState(core.DocStateParsed) + h.lock.StateChanged(0) + + // ReadAt(Parsed) must be admitted while the write is still active. + ra := h.startReadAt(context.Background(), core.DocStateParsed, []core.URI{doc.URI}) + ra.awaitEntered("ReadAt was not admitted during the build phase") + ra.finish() + + // A plain Read must still be blocked. + r := h.startRead(context.Background()) + r.assertBlocked("plain Read proceeded while write was active") + + w.finish() + r.awaitEntered("read did not proceed after write finished") + r.finish() } -// TestDowngradeAtomicity is the critical correctness test: it verifies that no new -// write can acquire the lock between the call to downgrade and the end of do. -// Without atomic downgrade this test would be racy. -func TestDowngradeAtomicity(t *testing.T) { - lock := NewDefaultLock() +// TestReadAtBlockedDuringMutationPhase verifies that ReadAt does not trust +// document states while a write is in its mutation phase, even if the bits are +// set: the write may be about to reset the document. +func TestReadAtBlockedDuringMutationPhase(t *testing.T) { + h := newLockHarness(t) + doc := h.doc("file:///a.test", core.DocStateParsed) - inReadPhase := make(chan struct{}) - releaseReadPhase := make(chan struct{}) + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") - // First write: downgrades and then lingers in the read phase. - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - downgrade() // transition to read phase - close(inReadPhase) - <-releaseReadPhase // hold the read lock - }) - }() - <-inReadPhase - - // Second write: must not start its do until the first write's do has returned. - var mu sync.Mutex - var events []string - record := func(s string) { - mu.Lock() - events = append(events, s) - mu.Unlock() - } + ra := h.startReadAt(context.Background(), core.DocStateParsed, []core.URI{doc.URI}) + ra.assertBlocked("ReadAt was admitted during the mutation phase") - write2Started := make(chan struct{}) - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - record("write2") - close(write2Started) - }) - }() + // Entering the build phase admits the ReadAt. + h.lock.StateChanged(0) + ra.awaitEntered("ReadAt was not admitted after the build phase began") + ra.finish() - // Give the second write time to queue up while the read phase is still held. - time.Sleep(shortWait) - record("release") - close(releaseReadPhase) + w.finish() +} - select { - case <-write2Started: - case <-time.After(longWait): - t.Fatal("second write never started") - } +// TestReadAtWaitsForRequestedState verifies that ReadAt blocks until the +// document reaches the requested states, then proceeds mid-write. +func TestReadAtWaitsForRequestedState(t *testing.T) { + h := newLockHarness(t) + doc := h.doc("file:///a.test", core.DocStateParsed) - mu.Lock() - defer mu.Unlock() - assert.Equal(t, []string{"release", "write2"}, events, - "write2 must not start before the read phase is released") -} + // ReadAt requires Linked, which the document does not have yet. + ra := h.startReadAt(context.Background(), core.DocStateParsed|core.DocStateLinked, []core.URI{doc.URI}) + ra.assertBlocked("ReadAt proceeded before the document was linked") -// TestWriteWaitsForActiveReaders verifies that Write only acquires the lock -// after all in-progress Read calls have completed. -func TestWriteWaitsForActiveReaders(t *testing.T) { - lock := NewDefaultLock() + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") + doc.SetState(doc.State().With(core.DocStateLinked)) + h.lock.StateChanged(0) - readerInside := make(chan struct{}) - releaseReader := make(chan struct{}) - go func() { - err := lock.Read(context.Background(), func(ctx context.Context) { - close(readerInside) - <-releaseReader - }) - assert.NoError(t, err) - }() - <-readerInside - - var mu sync.Mutex - var events []string - record := func(s string) { - mu.Lock() - events = append(events, s) - mu.Unlock() - } + ra.awaitEntered("ReadAt was not admitted after the document reached the state") + ra.finish() + w.finish() +} - writeStarted := make(chan struct{}) - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - record("write") - close(writeStarted) - }) - }() +// TestReadAtWaitsForUnknownDocument verifies that a URI with no document counts +// as not ready and that ReadAt is admitted once a build creates the document. +func TestReadAtWaitsForUnknownDocument(t *testing.T) { + h := newLockHarness(t) + uri := core.ParseURI("file:///new.test") - time.Sleep(shortWait) - record("release") - close(releaseReader) + ra := h.startReadAt(context.Background(), core.DocStateParsed, []core.URI{uri}) + ra.assertBlocked("ReadAt proceeded for an unknown document") - select { - case <-writeStarted: - case <-time.After(longWait): - t.Fatal("write never started after reader finished") - } + // A build cycle creates and parses the document. + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") + h.doc("file:///new.test", core.DocStateParsed) + h.lock.StateChanged(0) + w.finish() - mu.Lock() - defer mu.Unlock() - assert.Equal(t, []string{"release", "write"}, events, - "write must not start before the reader releases") + ra.awaitEntered("ReadAt was not admitted after the document was created and parsed") + ra.finish() } -// TestNewWriteCancelsPendingWrite verifies that when a second Write call arrives -// while the first is still waiting to acquire the lock, the first still runs do -// (so document mutations are never lost) but receives a cancelled context so it -// can skip expensive work such as building. -func TestNewWriteCancelsPendingWrite(t *testing.T) { - lock := NewDefaultLock() +// TestReadAtContextCancelledWhileWaiting verifies that a blocked ReadAt returns +// ctx.Err() when cancelled. +func TestReadAtContextCancelledWhileWaiting(t *testing.T) { + h := newLockHarness(t) + uri := core.ParseURI("file:///missing.test") - // Hold read lock so both writers must wait. - readerAcquired := make(chan struct{}) - releaseReader := make(chan struct{}) - go func() { - err := lock.Read(context.Background(), func(ctx context.Context) { - close(readerAcquired) - <-releaseReader - }) - assert.NoError(t, err) - }() - <-readerAcquired + ctx, cancel := context.WithCancel(context.Background()) + ra := h.startReadAt(ctx, core.DocStateParsed, []core.URI{uri}) + ra.assertBlocked("ReadAt proceeded for a missing document") - g1Done := make(chan struct{}) - var g1Ctx context.Context - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - g1Ctx = ctx - }) - close(g1Done) - }() - time.Sleep(shortWait) // let G1 queue up in cond.Wait + cancel() + ra.awaitDone("ReadAt did not return after context cancellation") + assert.ErrorIs(t, ra.err, context.Canceled) + ra.assertNotEntered("ReadAt callback ran despite cancellation") +} - g2Done := make(chan struct{}) - var g2Ctx context.Context - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - g2Ctx = ctx - }) - close(g2Done) - }() +// TestReadAtBlockedByPendingWrite verifies write priority for ReadAt: once a +// new write is pending, ReadAt is not admitted even if its documents are ready. +func TestReadAtBlockedByPendingWrite(t *testing.T) { + h := newLockHarness(t) + doc := h.doc("file:///a.test", 0) - close(releaseReader) + w1 := h.startWrite(context.Background()) + w1.awaitEntered("W1 was not admitted") + doc.SetState(core.DocStateParsed) + h.lock.StateChanged(0) - select { - case <-g1Done: - case <-time.After(longWait): - t.Fatal("G1 did not complete") - } - select { - case <-g2Done: - case <-time.After(longWait): - t.Fatal("G2 did not complete") - } + // Write 2 queues up behind Write 1. + w2 := h.startWrite(context.Background()) + time.Sleep(shortWait) // let W2 register as a waiter - // G1 must have run (mutations must not be lost) but with a cancelled context. - assert.ErrorIs(t, g1Ctx.Err(), context.Canceled, "G1 should have received a cancelled context") - // G2 is the newest writer and must receive a live context. - assert.NoError(t, g2Ctx.Err(), "G2 should have received a fresh context") -} + // ReadAt must be blocked despite the document being ready. + ra := h.startReadAt(context.Background(), core.DocStateParsed, []core.URI{doc.URI}) + ra.assertBlocked("ReadAt was admitted while a write was pending") -// TestNewWriteCancelsActiveWrite verifies that a second Write call cancels the -// first while its do callback is actively running, and then runs itself. -func TestNewWriteCancelsActiveWrite(t *testing.T) { - lock := NewDefaultLock() + w1.finish() + w2.awaitEntered("W2 was never admitted") + w2.finish() - firstRunning := make(chan struct{}) + ra.awaitEntered("ReadAt was not admitted after the pending write finished") + ra.finish() +} - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - close(firstRunning) - <-ctx.Done() // block until cancelled - }) - }() - <-firstRunning +// TestWriteWaitsForReadAtReaders verifies that a new write acquires the lock +// only after ReadAt readers admitted during the previous write have drained. +func TestWriteWaitsForReadAtReaders(t *testing.T) { + h := newLockHarness(t) + doc := h.doc("file:///a.test", 0) - // Second Write cancels the first and runs after it exits. - secondCalled := false - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - secondCalled = true - }) + w1 := h.startWrite(context.Background()) + w1.awaitEntered("W1 was not admitted") + doc.SetState(core.DocStateParsed) + h.lock.StateChanged(0) - assert.True(t, secondCalled) -} + // ReadAt is admitted mid-write and lingers. + ra := h.startReadAt(context.Background(), core.DocStateParsed, []core.URI{doc.URI}) + ra.awaitEntered("ReadAt was not admitted during the build phase") -// TestDowngradeIsIdempotent verifies that calling downgrade multiple times -// does not panic, deadlock, or corrupt the lock state. -func TestDowngradeIsIdempotent(t *testing.T) { - lock := NewDefaultLock() + w1.finish() // Write 1 finishes; the ReadAt reader is still active. - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - downgrade() - downgrade() - downgrade() - }) + w2 := h.startWrite(context.Background()) + w2.assertBlocked("W2 started before the ReadAt reader released") - // Lock should be fully released; a subsequent read must succeed. - done := make(chan struct{}) - go func() { - err := lock.Read(context.Background(), func(ctx context.Context) { close(done) }) - assert.NoError(t, err) - }() - select { - case <-done: - case <-time.After(longWait): - t.Fatal("lock was not released after idempotent downgrade calls") - } + ra.finish() + w2.awaitEntered("W2 was never admitted after the ReadAt reader released") + w2.finish() } -// TestWriteHasPriorityOverQueuedRead verifies the ordering: -// Write 1 enters -> Read queues up -> Write 2 arrives -> Write 2 runs before the Read. -func TestWriteHasPriorityOverQueuedRead(t *testing.T) { - lock := NewDefaultLock() +// TestReadAtStress hammers ReadAt against a document that is continuously +// reset and rebuilt. Run with -race: it verifies both the admission logic +// (data behind a requested state bit is always present) and the memory +// publication through the atomic document state. +func TestReadAtStress(t *testing.T) { + h := newLockHarness(t) - write1InDo := make(chan struct{}) - doDowngrade := make(chan struct{}) - write1Done := make(chan struct{}) + stable := h.doc("file:///stable.test", core.DocStateParsed) + stable.ParserErrors = []*core.ParserError{} - // Write 1: signal when in do, wait for downgrade cue, then linger in validation. - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - close(write1InDo) - <-doDowngrade - downgrade() - <-write1Done - }) - }() - <-write1InDo // Write 1 is now holding the write lock. + target := h.doc("file:///target.test", 0) - // Read: queues up while Write 1 holds the lock. - readDone := make(chan struct{}) - go func() { - err := lock.Read(context.Background(), func(ctx context.Context) { close(readDone) }) - assert.NoError(t, err) - }() - time.Sleep(shortWait) // let the Read block on readyCh. + stop := make(chan struct{}) + var wg sync.WaitGroup - // Write 2: arrives after the Read is already queued. - write2Done := make(chan struct{}) - var mu sync.Mutex - var events []string - record := func(s string) { - mu.Lock() - events = append(events, s) - mu.Unlock() + // Writer: continuously reset and rebuild the target document. + wg.Go(func() { + defer close(stop) + for range 300 { + h.lock.Write(context.Background(), func(ctx context.Context) { + // Mutation phase: reset. + target.SetState(0) + target.ParserErrors = nil + // Build phase: rebuild and publish. + target.ParserErrors = []*core.ParserError{} + target.SetState(core.DocStateParsed) + h.lock.StateChanged(0) + }) + } + }) + + // Readers: ReadAt must only ever observe fully published data. + for _, doc := range []*core.Document{target, stable} { + for range 2 { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + } + ctx, cancel := context.WithTimeout(context.Background(), longWait) + _ = h.lock.ReadAt(ctx, core.DocStateParsed, []core.URI{doc.URI}, func(ctx context.Context) { + if doc.ParserErrors == nil { + t.Error("ReadAt observed unpublished data: ParserErrors is nil despite Parsed state") + } + }) + cancel() + } + }) + } } - go func() { - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - record("write2") - }) - close(write2Done) - }() - time.Sleep(shortWait) // let Write 2 register as a waiter. - // Release Write 1 through downgrade and then finish validation. - close(doDowngrade) - time.Sleep(shortWait) // let Write 1 downgrade and Write 2 attempt to acquire. - close(write1Done) + wg.Wait() +} - select { - case <-write2Done: - case <-time.After(longWait): - t.Fatal("Write 2 did not complete") - } - select { - case <-readDone: - case <-time.After(longWait): - t.Fatal("Read did not complete") - } +// TestReadAtWorkspaceWide verifies the whole-workspace form (no URIs): it +// waits for the floor reported through StateChanged and is admitted mid-write +// once the floor covers the requested states. +func TestReadAtWorkspaceWide(t *testing.T) { + h := newLockHarness(t) - record("read") + // The floor starts at 0, so a workspace-wide ReadAt must block. + ra := h.startReadAt(context.Background(), core.DocStateParsed, nil) + ra.assertBlocked("workspace-wide ReadAt proceeded before any floor was reported") - mu.Lock() - defer mu.Unlock() - assert.Equal(t, []string{"write2", "read"}, events, - "Write 2 must run before the queued Read") -} + w := h.startWrite(context.Background()) + w.awaitEntered("write was not admitted") + h.lock.StateChanged(core.DocStateParsed | core.DocStateExportedSymbols) -// TestSafetyNetEnsuresDowngradeIsCalled verifies that the lock is fully released -// even when do never calls downgrade (e.g. it returns early on error). -func TestSafetyNetEnsuresDowngradeIsCalled(t *testing.T) { - lock := NewDefaultLock() + // Admitted mid-write: the floor covers Parsed while the write still runs. + ra.awaitEntered("workspace-wide ReadAt was not admitted after the floor was reported") + ra.finish() - lock.Write(context.Background(), func(ctx context.Context, downgrade func()) { - // Intentionally never call downgrade. - }) + // A higher state than the floor must still block. + linked := h.startReadAt(context.Background(), core.DocStateLinked, nil) + linked.assertBlocked("workspace-wide ReadAt proceeded beyond the reported floor") - // The lock must be fully released; a subsequent read must succeed. - done := make(chan struct{}) - go func() { - err := lock.Read(context.Background(), func(ctx context.Context) { close(done) }) - assert.NoError(t, err) - }() - select { - case <-done: - case <-time.After(longWait): - t.Fatal("lock was not released after do returned without calling downgrade") - } + h.lock.StateChanged(core.DocStateLinked) + linked.awaitEntered("workspace-wide ReadAt was not admitted after the floor was raised") + linked.finish() + + w.finish() +} + +// TestReadAtWorkspaceWideFloorResetsOnNewWrite verifies that the floor is +// cleared when a new write acquires the lock, so workspace-wide ReadAt calls +// wait for the new build cycle to re-establish it. +func TestReadAtWorkspaceWideFloorResetsOnNewWrite(t *testing.T) { + h := newLockHarness(t) + + // First write establishes a floor; after it ends the floor persists. + w1 := h.startWrite(context.Background()) + w1.awaitEntered("W1 was not admitted") + h.lock.StateChanged(core.DocStateParsed) + w1.finish() + err := h.lock.ReadAt(context.Background(), core.DocStateParsed, nil, func(ctx context.Context) {}) + assert.NoError(t, err, "floor must persist after the write ends") + + // Second write: the floor resets on acquisition (mutation phase). + w2 := h.startWrite(context.Background()) + w2.awaitEntered("W2 was not admitted") + + ra := h.startReadAt(context.Background(), core.DocStateParsed, nil) + ra.assertBlocked("workspace-wide ReadAt trusted a stale floor during a new write") + + // The write ends without re-establishing the floor: still blocked. + w2.finish() + ra.assertBlocked("workspace-wide ReadAt proceeded without a re-established floor") + + // A new build cycle re-establishes the floor. + w3 := h.startWrite(context.Background()) + w3.awaitEntered("W3 was not admitted") + h.lock.StateChanged(core.DocStateParsed) + ra.awaitEntered("workspace-wide ReadAt was not admitted after the floor was re-established") + ra.finish() + w3.finish() } diff --git a/workspace/services.go b/workspace/services.go index 4d4d0466..9872624c 100644 --- a/workspace/services.go +++ b/workspace/services.go @@ -38,7 +38,7 @@ func SetupDefaultServices(sc *service.Container) { service.Put(sc, NewDefaultIncludeFilter(sc)) } if !service.Has[Lock](sc) { - service.Put(sc, NewDefaultLock()) + service.Put(sc, NewDefaultLock(sc)) } if !service.Has[DocumentUpdater](sc) { service.Put(sc, NewDefaultDocumentUpdater(sc))