Files
ConfigKit/internal/engine/import_acquisition.go

542 lines
23 KiB
Go

package engine
import (
"bytes"
"errors"
"io/fs"
"strconv"
"gitea.maximumdirect.net/eric/ConfigKit/internal/document"
yamladapter "gitea.maximumdirect.net/eric/ConfigKit/internal/yaml"
)
type importAdmissionOutcome uint8
const (
importAdmissionReady importAdmissionOutcome = iota
importAdmissionOptionalMissing
)
type admittedImport struct {
canonical string
identity resourceIdentity
nodeKey importNodeKey
source SourceInfo
sourceIndex int
payload *document.Node
declarations []importDeclarationSpec
sequenceDeclarations []sequenceDeclarationSpec
processing *sourceVersionProcessingResult
}
type importAdmissionResult struct {
outcome importAdmissionOutcome
source admittedImport
}
func acquireImportedInput(request importAcquisitionRequest, policy *acquisitionPolicy, state *acquisitionState, tier *tierAcquisitionState) (*composedDocument, []SourceInfo, *resolvedFilesystemContext, []Diagnostic) {
source := request.input.input.source
context := request.input.input.filesystem
main := request.main
resolved, diagnostic := resolveFilesystemContext(source, context, main, tier.hostToken)
if diagnostic != nil {
return nil, sourceInfos(main.source), nil, []Diagnostic{*diagnostic}
}
if resolved.native == nil {
if diagnostic := validateResolvedFilesystemDirectories(resolved); diagnostic != nil {
return nil, sourceInfos(main.source), nil, []Diagnostic{*diagnostic}
}
}
access, err := openResolvedFilesystemAccess(resolved)
if err != nil {
diagnostic := globFilesystemDiagnostic(err, "configuration root could not be opened", Location{})
diagnostic.Primary = nil
return nil, sourceInfos(main.source), nil, []Diagnostic{diagnostic}
}
composition, sources, diagnostics := acquireImportedWithAccess(request, policy, state, tier, access, resolved)
if len(diagnostics) != 0 {
return composition, sources, nil, diagnostics
}
return composition, sources, &resolved, nil
}
func acquireImportedWithAccess(request importAcquisitionRequest, policy *acquisitionPolicy, loadState *acquisitionState, tier *tierAcquisitionState, access *filesystemAccess, resolved resolvedFilesystemContext) (*composedDocument, []SourceInfo, []Diagnostic) {
closed := false
closeAccess := func() error {
if closed {
return nil
}
closed = true
return access.close()
}
defer func() { _ = closeAccess() }()
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)
}
visited := tier.visited
if !access.contains(basePath) || (mainPath != "" && !access.contains(mainPath)) {
_ = closeAccess()
return nil, sourceInfos(main.source), []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "selected source and import base must stay inside the configuration root", nil)}
}
for _, directory := range []string{access.rootPath, basePath} {
info, err := access.stat(directory)
if err != nil || !info.IsDir() {
_ = closeAccess()
return nil, sourceInfos(main.source), []Diagnostic{importDiagnostic(CodeImportInvalid, "configuration root and import base must be directories", nil)}
}
}
mainInfo := cloneSourceInfo(*main.source)
mainInfo.Destination = destination.clone()
mainIsItem := main.source.placementMode == placementModeSequenceItem
if mainPath != "" {
mainInfo.Path = mainPath
}
mainResource := access.identity(mainPath)
mainNodeKey := importKeyFor(access, mainPath, mainPath == "")
if mainPath == "" {
mainResource = access.syntheticIdentity()
}
if previous, repeated := visited[mainResource]; repeated {
_ = closeAccess()
return nil, sourceInfos(main.source), []Diagnostic{importRepeatedDiagnostic(Location{SourceID: main.source.ID}, previous, importRoute{})}
}
accounting.sources++
accounting.bytes += main.bytes
accounting.nodes += countInputNodes(main.document)
directiveContext := request.input.directiveContext
if directiveContext == 0 {
directiveContext = rootDirectiveContextImportsOnly
}
if processing != nil && processing.directiveContext != 0 {
directiveContext = processing.directiveContext
}
request.input.directiveContext = directiveContext
state := importTraversal{
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 {
state.itemScope = true
state.itemPrefix = destination.clone()
state.itemFallback = destination.clone()
state.scope.destination = Path{}
state.sources[0] = cloneSourceInfo(mainInfo)
}
if diagnostic := state.checkInitialLimits(); diagnostic != nil {
_ = closeAccess()
return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
}
mainIdentity := mainPath
if mainIdentity == "" {
mainIdentity = "<selected-source>"
}
mainRoute := importRoute{canonical: mainIdentity}
state.importGraph.addNode(importGraphNode{key: mainNodeKey, canonical: mainIdentity, sourceID: mainInfo.ID, role: importRoleMain})
mainRoute.key = mainNodeKey
state.active[mainResource] = 0
tier.visited[mainResource] = mainRoute
state.stack = append(state.stack, mainRoute)
var body *document.Node
var declarations []importDeclarationSpec
var sequenceDeclarations []sequenceDeclarationSpec
var diagnostics []Diagnostic
if processing != nil {
body = processing.payload
declarations = processing.imports
sequenceDeclarations = cloneSequenceDeclarations(processing.sequences)
} else {
body, declarations, diagnostics = splitImportDirectives(main.document)
}
if len(diagnostics) == 0 {
if mainIsItem {
state.scope.contributions = append(state.scope.contributions, acquiredContribution{
source: cloneSourceInfo(mainInfo), identity: mainNodeKey, role: importRoleMain,
scopeOwner: mainNodeKey, payload: body, document: body, processing: processing,
sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations),
})
diagnostics = state.walkDeclarations(basePath, declarations, 0, Path{}, 0)
} else {
_, placed, err := placeDocument(body, destination, body.Location, maxDocumentDepth, maxDocumentNodes)
placementNodes := destinationWrapperNodeCount(destination)
if err != nil {
diagnostics = []Diagnostic{sourceDiagnostic(CodeSourceLimitExceeded, main.source.ID, "configuration exceeds the source depth or node limit after destination placement")}
} else if accounting.nodes+placementNodes > limits.nodes {
diagnostics = []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", nil)}
} else {
accounting.nodes += placementNodes
state.scope.contributions = append(state.scope.contributions, acquiredContribution{
source: cloneSourceInfo(mainInfo), payload: body, document: placed,
identity: mainNodeKey, role: importRoleMain, scopeOwner: mainNodeKey,
destination: destination.clone(), processing: processing,
sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations),
})
diagnostics = state.walkDeclarations(basePath, declarations, 0, destination, 0)
}
}
}
if len(diagnostics) == 0 {
scheduleAcquisitionScopes(&state.scope, indexImportDependencies(state.importGraph))
}
closeErr := closeAccess()
if len(diagnostics) != 0 {
if mainIsItem {
diagnostics = prependCompositionDiagnosticPaths(diagnostics, destination.Index(0))
}
return nil, cloneSourceInfos(state.sources), diagnostics
}
if closeErr != nil {
return nil, cloneSourceInfos(state.sources), []Diagnostic{importDiagnostic(CodeImportRead, "configuration root could not be closed", nil)}
}
acquisition := state.scope
if mainIsItem {
localScope := state.scope
acquisition = acquisitionScope{
owner: mainNodeKey, destination: destination.clone(),
contributions: []acquiredContribution{{
source: cloneSourceInfo(mainInfo), payload: body, identity: mainNodeKey,
role: importRoleMain, scopeOwner: mainNodeKey, destination: destination.clone(),
processing: processing, sequenceDeclarations: cloneSequenceDeclarations(sequenceDeclarations), pendingItem: &localScope,
}},
}
}
if request.input.deferAssembly {
if diagnostic := reservePendingItemPlacements(&acquisition, limits, accounting); diagnostic != nil {
return nil, cloneSourceInfos(state.sources), []Diagnostic{*diagnostic}
}
return &composedDocument{
acquisition: []acquisitionScope{cloneAcquisitionScope(acquisition)},
importGraph: state.importGraph.clone(), inputBytes: accounting.bytes, inputNodes: accounting.nodes,
}, cloneSourceInfos(state.sources), nil
}
acquired := &composedDocument{
acquisition: []acquisitionScope{cloneAcquisitionScope(acquisition)},
importGraph: state.importGraph.clone(), inputBytes: accounting.bytes, inputNodes: accounting.nodes,
}
input := assembledInput{
parsed: main, composition: acquired, acquisition: cloneAcquisitionScopes(acquired.acquisition),
sources: state.sources, versionState: versionState, accounting: accounting,
}
composed, diagnostics := composeAcquiredInputWithPolicies(input, collectionPolicyScope{}, policy.sequencePolicy)
return composed, cloneSourceInfos(state.sources), diagnostics
}
func (s *importTraversal) checkInitialLimits() *Diagnostic {
if s.state.accounting.sources > s.policy.importLimits.sources {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source-count limit", nil)
return &value
}
if s.state.accounting.bytes > s.policy.importLimits.bytes {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate byte limit", nil)
return &value
}
if s.state.accounting.nodes > s.policy.importLimits.nodes {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", nil)
return &value
}
return nil
}
func (s *importTraversal) walkDeclarations(basePath string, declarations []importDeclarationSpec, depth int, parentDestination Path, sourceIndex int) []Diagnostic {
policies := make([]resolvedImportPolicy, len(declarations))
for index, spec := range declarations {
policy, diagnostics := resolveImportPolicy(spec, s.policy.importDefaults)
if len(diagnostics) != 0 {
return diagnostics
}
if policy.placement == placementModeSequenceItem && len(appendDestination(parentDestination, policy.destination).segments) == 0 {
site := spec.fieldSites["placement"]
return []Diagnostic{{Code: CodeImportInvalid, Severity: SeverityError, Path: site.path.clone(), Message: "sequence-item placement requires a nonempty destination", Primary: cloneLocation(&site.location)}}
}
policies[index] = policy
}
for index, spec := range declarations {
policy := policies[index]
var selection importSelection
var diagnostics []Diagnostic
switch spec.selector {
case importSelectorPath:
selection, diagnostics = selectNamedImportCandidates(s.access, basePath, spec, policy)
case importSelectorDirectory:
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.policy.importLimits, s.state.accounting)
default:
location := spec.location
return []Diagnostic{importDiagnostic(CodeImportInvalid, "import selector is invalid", &location)}
}
if len(diagnostics) != 0 {
return diagnostics
}
childDestination := appendDestination(parentDestination, policy.destination)
// Count admitted records rather than lexical candidates: optional missing
// paths are omitted during admission, and a partially present list is not empty.
sourcesBefore := len(s.sources)
for _, candidate := range selection.candidates {
if diagnostics := s.visitPath(candidate, spec, policy, childDestination, policy.destination, depth+1); len(diagnostics) != 0 {
return diagnostics
}
}
if !policy.required && len(s.sources) == sourcesBefore {
s.recordEmptyImportSelection(sourceIndex, spec)
}
}
return nil
}
func (s *importTraversal) recordEmptyImportSelection(sourceIndex int, spec importDeclarationSpec) {
// The declaring source's position survives descendant appends and optional
// missing-source rollback, unlike a pointer into the source slice.
s.sources[sourceIndex].importSelections = append(s.sources[sourceIndex].importSelections, importSelectionEvidence{
path: spec.path.clone(), location: spec.location,
selector: spec.selector, outcome: importSelectionNone,
})
}
func (s *importTraversal) admitImportSource(canonical string, identity resourceIdentity, nodeKey importNodeKey, candidate selectedImportCandidate, policy resolvedImportPolicy, destination Path, location Location) (importAdmissionResult, []Diagnostic) {
sourceID := "source-" + strconv.Itoa(s.state.accounting.sources+1)
kind := SourceKindFile
label := candidate.label
if label == "" {
label = candidate.path
}
if s.access.native != nil {
kind = SourceKindFilesystem
label = s.access.label + "/" + canonical
}
info := SourceInfo{ID: sourceID, Label: label, Path: canonical, Destination: s.sourceDestination(destination), Layer: LayerBase, Kind: kind, ImportedFrom: &location}
if policy.placement == placementModeSequenceItem {
info.placementMode = placementModeSequenceItem
}
// A source's own item placement still has an exact sequence-container
// destination. Only an unresolved enclosing item index makes the path a
// fallback container for this snapshot.
info.destinationUnresolved = s.itemScope
if kind == SourceKindFilesystem {
info.namespace = s.access.token
}
sourceIndex := len(s.sources)
s.sources = append(s.sources, info)
data, skipped, diagnostic := s.readImport(canonical, sourceID, policy.required)
if diagnostic != nil {
return importAdmissionResult{}, []Diagnostic{*diagnostic}
}
if skipped {
s.sources = s.sources[:len(s.sources)-1]
return importAdmissionResult{outcome: importAdmissionOptionalMissing}, nil
}
s.state.accounting.sources++
root, issues := yamladapter.Parse(bytes.NewReader(data), sourceID, yamladapter.Limits{
Depth: maxDocumentDepth, Nodes: maxDocumentNodes,
})
if len(issues) != 0 {
return importAdmissionResult{}, parserDiagnostics(sourceID, issues)
}
nodeCount := countInputNodes(root)
if s.state.accounting.nodes+nodeCount > s.policy.importLimits.nodes {
value := importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", &location)
return importAdmissionResult{}, []Diagnostic{value}
}
s.state.accounting.nodes += nodeCount
directiveContext := s.request.input.directiveContext
if directiveContext == 0 {
directiveContext = rootDirectiveContextImportsOnly
}
versionDestination := destination
if s.itemScope {
versionDestination = appendDestination(s.itemPrefix.Index(0), destination)
}
var processing sourceVersionProcessingResult
if s.state.versionState != nil && s.state.versionState.mode != sourceVersionModeDefault {
processed, failures := s.state.versionState.processWithDirectiveContext(root, sourceVersionContribution, versionDestination, sourceID, policy.placement, directiveContext)
if len(failures) != 0 {
return importAdmissionResult{}, failures
}
processing = processed
} else {
processed := processSourceVersionWithDirectiveContext(root, s.request.input.protectionRule, sourceVersionContribution, versionDestination, sourceVersionModeDefault, directiveContext)
if len(processed.diagnostics) != 0 {
return importAdmissionResult{}, processed.diagnostics
}
processing = processed
}
return importAdmissionResult{source: admittedImport{
canonical: canonical, identity: identity, nodeKey: nodeKey,
source: cloneSourceInfo(info), sourceIndex: sourceIndex, payload: processing.payload,
declarations: processing.imports,
sequenceDeclarations: cloneSequenceDeclarations(processing.sequences),
processing: &processing,
}}, nil
}
func (s *importTraversal) visitPath(candidate selectedImportCandidate, spec importDeclarationSpec, policy resolvedImportPolicy, destination, relativeDestination Path, depth int) []Diagnostic {
location := candidate.site.location
if location.SourceID == "" && location.Line == 0 && location.Column == 0 {
location = spec.location
}
if !s.access.contains(candidate.accessPath) {
return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target is outside the configuration root", &location)}
}
canonical, err := s.access.canonical(candidate.accessPath)
if err != nil {
if !policy.required && errors.Is(err, fs.ErrNotExist) {
return nil
}
code := CodeImportInvalid
message := "import target could not be canonicalized"
if errors.Is(err, fs.ErrNotExist) {
code, message = CodeImportMissing, "required import target does not exist"
} else if errors.Is(err, fs.ErrPermission) {
message = "import target cannot be accessed"
}
return []Diagnostic{importDiagnostic(code, message, &location)}
}
if !s.access.contains(canonical) {
return []Diagnostic{importDiagnostic(CodeImportOutsideRoot, "import target resolves outside the configuration root", &location)}
}
if depth > s.policy.importLimits.depth {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the active import-depth limit", &location)}
}
currentRoute := s.currentRoute(location)
identity := s.access.identity(canonical)
nodeKey := importKeyFor(s.access, canonical, false)
importerKey := s.currentImportNode()
if activeIndex, ok := s.active[identity]; ok {
closingEdge := s.importGraph.addEdge(importerKey, nodeKey, location)
if diagnostic, ok := s.importGraph.scheduleCycle(closingEdge, location); ok {
return []Diagnostic{diagnostic}
}
cycle := append([]importRoute(nil), s.stack[activeIndex:]...)
return []Diagnostic{importCycleDiagnostic(canonical, location, cycle)}
}
if previous, ok := s.tier.visited[identity]; ok {
s.importGraph.addEdge(importerKey, nodeKey, location)
return []Diagnostic{importRepeatedDiagnostic(location, previous, currentRoute)}
}
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)}
}
admission, diagnostics := s.admitImportSource(canonical, identity, nodeKey, candidate, policy, destination, location)
if len(diagnostics) != 0 {
return diagnostics
}
if admission.outcome == importAdmissionOptionalMissing {
return nil
}
admitted := admission.source
placementDestination := destination
mode := placementModeMapping
if policy.placement == placementModeSequenceItem {
mode = placementModeSequenceItem
}
role := importRoleMapping
if mode == placementModeSequenceItem {
role = importRoleItem
}
s.importGraph.addNode(importGraphNode{key: admitted.nodeKey, canonical: admitted.canonical, sourceID: admitted.source.ID, role: role})
s.importGraph.addEdge(importerKey, admitted.nodeKey, location)
if mode == placementModeSequenceItem {
route := importRoute{canonical: admitted.canonical, locations: currentRoute.locations, key: admitted.nodeKey}
s.active[admitted.identity] = len(s.stack)
s.tier.visited[admitted.identity] = route
s.stack = append(s.stack, route)
diagnostics := s.visitItemSource(admitted, destination, relativeDestination, depth)
delete(s.active, admitted.identity)
s.stack = s.stack[:len(s.stack)-1]
return diagnostics
}
_, placed, err := placeDocument(admitted.payload, placementDestination, admitted.payload.Location, maxDocumentDepth, maxDocumentNodes)
if err != nil {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the source depth or node limit after destination placement", &location)}
}
placementNodes := destinationWrapperNodeCount(placementDestination)
if s.state.accounting.nodes+placementNodes > s.policy.importLimits.nodes {
return []Diagnostic{importDiagnostic(CodeImportLimitExceeded, "configuration exceeds the aggregate node limit", &location)}
}
s.state.accounting.nodes += placementNodes
route := importRoute{canonical: admitted.canonical, locations: currentRoute.locations, key: admitted.nodeKey}
s.scope.contributions = append(s.scope.contributions, acquiredContribution{
source: cloneSourceInfo(admitted.source), payload: admitted.payload, document: placed,
identity: admitted.nodeKey, role: role, scopeOwner: s.scope.owner,
destination: placementDestination.clone(), relativeDestination: relativeDestination.clone(), processing: admitted.processing,
sequenceDeclarations: cloneSequenceDeclarations(admitted.sequenceDeclarations),
})
s.active[admitted.identity] = len(s.stack)
s.tier.visited[admitted.identity] = route
s.stack = append(s.stack, route)
diagnostics = s.walkDeclarations(s.access.dir(admitted.canonical), admitted.declarations, depth, destination, admitted.sourceIndex)
delete(s.active, admitted.identity)
s.stack = s.stack[:len(s.stack)-1]
return diagnostics
}
func (s *importTraversal) currentImportNode() importNodeKey {
if len(s.stack) != 0 {
return s.stack[len(s.stack)-1].key
}
key := s.scope.owner
if _, exists := s.importGraph.index[key]; !exists {
s.importGraph.addNode(importGraphNode{key: key, canonical: "<selected-source>", role: importRoleMain})
}
return key
}
func (s *importTraversal) sourceDestination(destination Path) Path {
if !s.itemScope {
return destination.clone()
}
// Until the enclosing item receives its assigned index, only the collection
// container is exact. Keep the local suffix in the acquisition projection.
return s.itemFallback.clone()
}
func (s *importTraversal) readImport(name, sourceID string, required bool) ([]byte, bool, *Diagnostic) {
remaining := s.policy.importLimits.bytes - s.state.accounting.bytes
limit := min(int64(maxSourceBytes), remaining)
if limit < 0 {
limit = 0
}
data, failure := readRegularFile(name, s.access.stat, s.access.open, limit)
if failure != nil {
if !required && failure.missingTarget {
return nil, true, nil
}
value := sourceReadDiagnostic(failure, sourceID, CodeImportLimitExceeded, "configuration exceeds a source or aggregate byte limit")
return nil, false, &value
}
s.state.accounting.bytes += int64(len(data))
return data, false, nil
}
func (s *importTraversal) currentRoute(location Location) importRoute {
locations := make([]Location, 0, len(s.stack)+1)
for _, route := range s.stack {
if len(route.locations) != 0 {
locations = append(locations, route.locations[len(route.locations)-1])
}
}
locations = append(locations, location)
return importRoute{locations: locations}
}