Consolidate import request ownership

This commit is contained in:
2026-10-25 13:02:46 +00:00
parent d05426d8a2
commit bd51b9c3f2
14 changed files with 178 additions and 157 deletions

View File

@@ -46,13 +46,11 @@ type importAccounting struct {
}
type importTraversal struct {
access *filesystemAccess
accounting *importAccounting
limits importLimits
rule *VersionRule
versionState *sourceVersionState
directiveContext rootDirectiveContext
defaults importDefaults
access *filesystemAccess
request importAcquisitionRequest
policy *acquisitionPolicy
state *acquisitionState
tier *tierAcquisitionState
scope acquisitionScope
itemScope bool
@@ -60,38 +58,37 @@ type importTraversal struct {
itemFallback Path
sources []SourceInfo
active map[resourceIdentity]int
visited map[resourceIdentity]importRoute
stack []importRoute
importGraph importInclusionGraph
}
func loadImportedDocuments(source Source, context *FilesystemContext, main parsedSource, rule *VersionRule, limits importLimits) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedDocumentsTracked(source, context, main, Path{}, rule, limits, &importAccounting{}, nil, nil)
}
func loadImportedDocumentsTracked(source Source, context *FilesystemContext, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, hostToken *namespaceToken) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedDocumentsTrackedWithVersion(source, context, main, destination, rule, limits, accounting, visited, hostToken, nil)
}
func loadImportedDocumentsTrackedWithVersion(source Source, context *FilesystemContext, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, hostToken *namespaceToken, versionState *sourceVersionState) (*composedDocument, []SourceInfo, []Diagnostic) {
var processing *sourceVersionProcessingResult
if versionState != nil && versionState.mode != sourceVersionModeDefault {
processing = versionState.mainProcessing
func compositionTraversal(policy sequencePolicy, rule *VersionRule, versionState *sourceVersionState, accounting *importAccounting, graph importInclusionGraph) importTraversal {
return importTraversal{
request: importAcquisitionRequest{input: inputAcquisitionRequest{protectionRule: rule}},
policy: &acquisitionPolicy{sequencePolicy: policy, importLimits: defaultImportLimits()},
state: &acquisitionState{accounting: accounting, versionState: versionState},
tier: &tierAcquisitionState{},
importGraph: graph.clone(),
}
return loadImportedDocumentsTrackedWithProcessing(source, context, main, destination, rule, limits, accounting, visited, hostToken, versionState, processing, sequencePolicy{})
}
func loadImportedDocumentsTrackedWithProcessing(source Source, context *FilesystemContext, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, hostToken *namespaceToken, versionState *sourceVersionState, processing *sourceVersionProcessingResult, policy sequencePolicy) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedDocumentsTrackedWithProcessingDeferred(source, context, main, destination, rule, limits, accounting, visited, hostToken, versionState, processing, policy, false)
}
func loadImportedDocumentsTrackedWithProcessingDeferred(source Source, context *FilesystemContext, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, hostToken *namespaceToken, versionState *sourceVersionState, processing *sourceVersionProcessingResult, policy sequencePolicy, deferAssembly bool) (*composedDocument, []SourceInfo, []Diagnostic) {
func acquireImportedInput(request importAcquisitionRequest, policy *acquisitionPolicy, state *acquisitionState, tier *tierAcquisitionState) (*composedDocument, []SourceInfo, []Diagnostic) {
source := request.input.input.source
context := request.input.input.filesystem
main := request.main
if nativeContext(source, context) {
access, basePath, mainPath, diagnostic := nativeFilesystemAccess(source, context)
if diagnostic != nil {
return nil, sourceInfos(main.source), []Diagnostic{*diagnostic}
}
return loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy, deferAssembly)
native := source.namespace
if native == nil && context != nil && context.FS != nil {
native = context.FS.namespace
}
return acquireImportedWithAccess(request, policy, state, tier, access, resolvedFilesystemContext{
native: native, hostToken: tier.hostToken, rootPath: access.rootPath,
basePath: basePath, mainPath: mainPath,
})
}
rootPath, basePath, mainPath, diagnostic := importFilesystem(source, context, main)
if diagnostic != nil {
@@ -103,37 +100,26 @@ func loadImportedDocumentsTrackedWithProcessingDeferred(source Source, context *
diagnostic.Primary = nil
return nil, sourceInfos(main.source), []Diagnostic{diagnostic}
}
if hostToken != nil {
access.token = hostToken
if tier.hostToken != nil {
access.token = tier.hostToken
}
return loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy, deferAssembly)
return acquireImportedWithAccess(request, policy, state, tier, access, resolvedFilesystemContext{
hostToken: tier.hostToken, rootPath: rootPath, basePath: basePath, mainPath: mainPath,
})
}
func loadImportedWithAccess(access *filesystemAccess, basePath, mainPath string, main parsedSource, rule *VersionRule, limits importLimits, accounting *importAccounting) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedWithAccessTracked(access, basePath, mainPath, main, Path{}, rule, limits, accounting, nil)
}
func loadImportedWithAccessTracked(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedWithAccessTrackedWithVersion(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, nil)
}
func loadImportedWithAccessTrackedWithVersion(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, versionState *sourceVersionState) (*composedDocument, []SourceInfo, []Diagnostic) {
var processing *sourceVersionProcessingResult
if versionState != nil && versionState.mode != sourceVersionModeDefault {
processing = versionState.mainProcessing
func acquireImportedWithAccess(request importAcquisitionRequest, policy *acquisitionPolicy, loadState *acquisitionState, tier *tierAcquisitionState, access *filesystemAccess, resolved resolvedFilesystemContext) (*composedDocument, []SourceInfo, []Diagnostic) {
main := request.main
destination := request.input.input.destination
limits := policy.importLimits
accounting := loadState.accounting
versionState := loadState.versionState
processing := request.processing
basePath, mainPath := resolved.basePath, resolved.mainPath
if tier.visited == nil {
tier.visited = make(map[resourceIdentity]importRoute)
}
return loadImportedWithAccessTrackedAndProcessing(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing)
}
func loadImportedWithAccessTrackedAndProcessing(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, versionState *sourceVersionState, processing *sourceVersionProcessingResult) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedWithAccessTrackedAndProcessingPolicy(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, sequencePolicy{})
}
func loadImportedWithAccessTrackedAndProcessingPolicy(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, versionState *sourceVersionState, processing *sourceVersionProcessingResult, policy sequencePolicy) (*composedDocument, []SourceInfo, []Diagnostic) {
return loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access, basePath, mainPath, main, destination, rule, limits, accounting, visited, versionState, processing, policy, false)
}
func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access *filesystemAccess, basePath, mainPath string, main parsedSource, destination Path, rule *VersionRule, limits importLimits, accounting *importAccounting, visited map[resourceIdentity]importRoute, versionState *sourceVersionState, processing *sourceVersionProcessingResult, policy sequencePolicy, deferAssembly bool) (*composedDocument, []SourceInfo, []Diagnostic) {
visited := tier.visited
if !access.contains(basePath) || (mainPath != "" && !access.contains(mainPath)) {
_ = access.close()
return nil, sourceInfos(main.source), []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "selected source and import base must stay inside the configuration root", nil)}
@@ -163,21 +149,19 @@ func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access *filesystem
accounting.sources++
accounting.bytes += main.bytes
accounting.nodes += countInputNodes(main.document)
defaults := importDefaults{}
if versionState != nil {
defaults = snapshotImportDefaults(versionState.importDefaults)
directiveContext := request.input.directiveContext
if directiveContext == 0 {
directiveContext = rootDirectiveContextImportsOnly
}
directiveContext := rootDirectiveContextImportsOnly
if processing != nil && processing.directiveContext != 0 {
directiveContext = processing.directiveContext
}
request.input.directiveContext = directiveContext
state := importTraversal{
access: access, accounting: accounting, limits: limits, rule: rule, versionState: versionState,
directiveContext: directiveContext,
defaults: defaults,
scope: acquisitionScope{owner: mainNodeKey, destination: destination.clone()},
sources: []SourceInfo{mainInfo},
active: make(map[resourceIdentity]int), visited: visited,
access: access, request: request, policy: policy, state: loadState, tier: tier,
scope: acquisitionScope{owner: mainNodeKey, destination: destination.clone()},
sources: []SourceInfo{mainInfo},
active: make(map[resourceIdentity]int),
importGraph: newImportInclusionGraph(),
}
if mainIsItem {
@@ -187,9 +171,7 @@ func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access *filesystem
state.scope.destination = Path{}
state.sources[0] = cloneSourceInfo(mainInfo)
}
if state.visited == nil {
state.visited = make(map[resourceIdentity]importRoute)
}
if diagnostic := state.checkInitialLimits(); diagnostic != nil {
_ = access.close()
return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
@@ -203,7 +185,7 @@ func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access *filesystem
state.importGraph.addNode(importGraphNode{key: mainNodeKey, canonical: mainIdentity, sourceID: mainInfo.ID, role: importRoleMain})
mainRoute.key = mainNodeKey
state.active[mainResource] = 0
state.visited[mainResource] = mainRoute
tier.visited[mainResource] = mainRoute
state.stack = append(state.stack, mainRoute)
var body *document.Node
var declarations []importDeclarationSpec
@@ -269,7 +251,7 @@ func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access *filesystem
}},
}
}
if deferAssembly {
if request.input.deferAssembly {
if diagnostic := reservePendingItemPlacements(&acquisition, limits, accounting); diagnostic != nil {
return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
}
@@ -286,7 +268,7 @@ func loadImportedWithAccessTrackedAndProcessingPolicyDeferred(access *filesystem
parsed: main, composition: acquired, acquisition: cloneAcquisitionScopes(acquired.acquisition),
sources: state.sources, versionState: versionState, accounting: accounting,
}
composed, diagnostics := composeAcquiredInputWithPolicies(input, collectionPolicyScope{}, policy)
composed, diagnostics := composeAcquiredInputWithPolicies(input, collectionPolicyScope{}, policy.sequencePolicy)
return composed, cloneSourceInfos(state.sources), diagnostics
}
@@ -311,11 +293,7 @@ func composeAcquiredInputWithPolicies(input assembledInput, lower collectionPoli
if accounting == nil {
accounting = &importAccounting{sources: len(input.sources), bytes: acquired.inputBytes, nodes: acquired.inputNodes}
}
traversal := importTraversal{
accounting: accounting,
limits: defaultImportLimits(), rule: rule, versionState: input.versionState,
importGraph: acquired.importGraph.clone(),
}
traversal := compositionTraversal(policy, rule, input.versionState, accounting, acquired.importGraph)
if len(scope.contributions) == 1 && scope.contributions[0].role == importRoleMain && scope.contributions[0].pendingItem != nil {
main := scope.contributions[0]
localScope := cloneAcquisitionScope(*main.pendingItem)
@@ -361,7 +339,7 @@ func composeAcquiredInputWithPolicies(input assembledInput, lower collectionPoli
}
if !main.placementCharged {
placementNodes := destinationWrapperNodeCountForMode(main.destination, placementModeSequenceItem)
if accounting.nodes+placementNodes > traversal.limits.nodes {
if accounting.nodes+placementNodes > traversal.policy.importLimits.nodes {
return nil, []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", main.source.ImportedFrom)}
}
accounting.nodes += placementNodes
@@ -401,10 +379,7 @@ func composeAcquiredScopeWithPolicy(
versionState *sourceVersionState,
accounting *importAccounting,
) (*composedDocument, []Diagnostic) {
traversal := importTraversal{
accounting: accounting, limits: defaultImportLimits(), rule: rule,
versionState: versionState, importGraph: graph.clone(),
}
traversal := compositionTraversal(policy, rule, versionState, accounting, graph)
if diagnostics := traversal.composePendingItems(scope); len(diagnostics) != 0 {
return nil, diagnostics
}
@@ -454,15 +429,15 @@ func reservePendingItemPlacements(scope *acquisitionScope, limits importLimits,
}
func (s *importTraversal) checkInitialLimits() *Diagnostic {
if s.accounting.sources > s.limits.sources {
if s.state.accounting.sources > s.policy.importLimits.sources {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source-count limit", nil)
return &value
}
if s.accounting.bytes > s.limits.bytes {
if s.state.accounting.bytes > s.policy.importLimits.bytes {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate byte limit", nil)
return &value
}
if s.accounting.nodes > s.limits.nodes {
if s.state.accounting.nodes > s.policy.importLimits.nodes {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", nil)
return &value
}
@@ -472,7 +447,7 @@ func (s *importTraversal) checkInitialLimits() *Diagnostic {
func (s *importTraversal) walkDeclarations(basePath string, declarations []importDeclarationSpec, depth int, parentDestination Path) []Diagnostic {
policies := make([]resolvedImportPolicy, len(declarations))
for index, spec := range declarations {
policy, diagnostics := resolveImportPolicy(spec, s.defaults)
policy, diagnostics := resolveImportPolicy(spec, s.policy.importDefaults)
if len(diagnostics) != 0 {
return diagnostics
}
@@ -490,9 +465,9 @@ func (s *importTraversal) walkDeclarations(basePath string, declarations []impor
case importSelectorPath:
selection, diagnostics = selectNamedImportCandidates(s.access, basePath, spec, policy)
case importSelectorDirectory:
selection, diagnostics = selectDirectoryImportCandidates(s.access, basePath, spec, policy, s.limits, s.accounting)
selection, diagnostics = selectDirectoryImportCandidates(s.access, basePath, spec, policy, s.policy.importLimits, s.state.accounting)
case importSelectorGlob:
selection, diagnostics = selectGlobImportCandidates(s.access, basePath, spec, policy, s.limits, s.accounting)
selection, diagnostics = selectGlobImportCandidates(s.access, basePath, spec, policy, s.policy.importLimits, s.state.accounting)
default:
location := spec.location
return []Diagnostic{importDiagnostic(CodeImportInvalid, "import selector is invalid", &location)}
@@ -554,7 +529,7 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
if !s.access.contains(canonical) {
return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target resolves outside the configuration root", &location)}
}
if depth > s.limits.depth {
if depth > s.policy.importLimits.depth {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the active import-depth limit", &location)}
}
@@ -570,18 +545,18 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
cycle := append([]importRoute(nil), s.stack[activeIndex:]...)
return []Diagnostic{importCycleDiagnostic(canonical, location, cycle)}
}
if previous, ok := s.visited[identity]; ok {
if previous, ok := s.tier.visited[identity]; ok {
s.importGraph.addEdge(importerKey, nodeKey, location)
return []Diagnostic{importRepeatedDiagnostic(location, previous, currentRoute)}
}
if s.accounting.sources+1 > s.limits.sources {
if s.state.accounting.sources+1 > s.policy.importLimits.sources {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source-count limit", &location)}
}
if _, err := s.access.relative(canonical); err != nil {
return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target could not be addressed inside the configuration root", &location)}
}
sourceID := "source-" + strconv.Itoa(s.accounting.sources+1)
sourceID := "source-" + strconv.Itoa(s.state.accounting.sources+1)
kind := SourceKindFile
label := candidate.label
if label == "" {
@@ -611,7 +586,7 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
s.sources = s.sources[:len(s.sources)-1]
return nil
}
s.accounting.sources++
s.state.accounting.sources++
root, issues := yamladapter.Parse(bytes.NewReader(data), sourceID, yamladapter.Limits{
Depth: maxDocumentDepth, Nodes: maxDocumentNodes,
@@ -620,26 +595,26 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
return parserDiagnostics(sourceID, issues)
}
nodeCount := countInputNodes(root)
if s.accounting.nodes+nodeCount > s.limits.nodes {
if s.state.accounting.nodes+nodeCount > s.policy.importLimits.nodes {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", &location)}
}
s.accounting.nodes += nodeCount
s.state.accounting.nodes += nodeCount
var body *document.Node
var declarations []importDeclarationSpec
var sequenceDeclarations []sequenceDeclarationSpec
var diagnostics []Diagnostic
var processing *sourceVersionProcessingResult
directiveContext := s.directiveContext
directiveContext := s.request.input.directiveContext
if directiveContext == 0 {
directiveContext = rootDirectiveContextImportsOnly
}
if s.versionState != nil && s.versionState.mode != sourceVersionModeDefault {
if s.state.versionState != nil && s.state.versionState.mode != sourceVersionModeDefault {
versionDestination := destination
if s.itemScope {
versionDestination = appendDestination(s.itemPrefix.Index(0), destination)
}
processed, failures := s.versionState.processWithDirectiveContext(root, sourceVersionContribution, versionDestination, sourceID, policy.placement, directiveContext)
processed, failures := s.state.versionState.processWithDirectiveContext(root, sourceVersionContribution, versionDestination, sourceID, policy.placement, directiveContext)
if len(failures) != 0 {
return failures
}
@@ -650,7 +625,7 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
if s.itemScope {
versionDestination = appendDestination(s.itemPrefix.Index(0), destination)
}
processed := processSourceVersionWithDirectiveContext(root, s.rule, sourceVersionContribution, versionDestination, sourceVersionModeDefault, directiveContext)
processed := processSourceVersionWithDirectiveContext(root, s.request.input.protectionRule, sourceVersionContribution, versionDestination, sourceVersionModeDefault, directiveContext)
if len(processed.diagnostics) != 0 {
return processed.diagnostics
}
@@ -672,7 +647,7 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
if mode == placementModeSequenceItem {
route := importRoute{canonical: canonical, locations: currentRoute.locations, key: nodeKey}
s.active[identity] = len(s.stack)
s.visited[identity] = route
s.tier.visited[identity] = route
s.stack = append(s.stack, route)
diagnostics := s.visitItemSource(canonical, nodeKey, destination, relativeDestination, depth, info, body, declarations, sequenceDeclarations, processing)
delete(s.active, identity)
@@ -684,10 +659,10 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source depth or node limit after destination placement", &location)}
}
placementNodes := destinationWrapperNodeCount(placementDestination)
if s.accounting.nodes+placementNodes > s.limits.nodes {
if s.state.accounting.nodes+placementNodes > s.policy.importLimits.nodes {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", &location)}
}
s.accounting.nodes += placementNodes
s.state.accounting.nodes += placementNodes
route := importRoute{canonical: canonical, locations: currentRoute.locations, key: nodeKey}
s.scope.contributions = append(s.scope.contributions, acquiredContribution{
source: cloneSourceInfo(info), payload: body, document: placed,
@@ -696,7 +671,7 @@ func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec impo
sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations),
})
s.active[identity] = len(s.stack)
s.visited[identity] = route
s.tier.visited[identity] = route
s.stack = append(s.stack, route)
diagnostics = s.walkDeclarations(s.access.dir(canonical), declarations, depth, destination)
delete(s.active, identity)
@@ -799,10 +774,10 @@ func (s *importTraversal) composePendingItems(scope *acquisitionScope) []Diagnos
}
placementNodes := destinationWrapperNodeCountForMode(contribution.destination, placementModeSequenceItem)
if !contribution.placementCharged {
if s.accounting.nodes+placementNodes > s.limits.nodes {
if s.state.accounting.nodes+placementNodes > s.policy.importLimits.nodes {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", contribution.source.ImportedFrom)}
}
s.accounting.nodes += placementNodes
s.state.accounting.nodes += placementNodes
}
contribution.payload = completed.root
contribution.document = placed
@@ -1126,7 +1101,7 @@ func projectSourceVersionEvidence(processing *sourceVersionProcessingResult, des
}
func (s *importTraversal) readImport(name, sourceID string, required bool) ([]byte, bool, *Diagnostic) {
remaining := s.limits.bytes - s.accounting.bytes
remaining := s.policy.importLimits.bytes - s.state.accounting.bytes
limit := min(int64(maxSourceBytes), remaining)
if limit < 0 {
limit = 0
@@ -1139,7 +1114,7 @@ func (s *importTraversal) readImport(name, sourceID string, required bool) ([]by
value := sourceReadDiagnostic(failure, sourceID, CodeImportLimitExceeded, "configuration exceeds a source or aggregate byte limit")
return nil, false, &value
}
s.accounting.bytes += int64(len(data))
s.state.accounting.bytes += int64(len(data))
return data, false, nil
}