Merge pull request #2585 from keboola/mvasko/PSGO-233/fix-orphaned-scheduler-manifest

fix(manifest): handle orphaned scheduler/orchestrator without panicking

Author: Matovidlo

internal/pkg/project/manifest/manifest.go

@@ -24,6 +24,20 @@ const (
 	BranchIDOverrideENV  = "KBC_BRANCH_ID"
 )
 
+// loadHintKey is a context key carrying an additional hint to log alongside the
+// "manifest loaded with warnings" warning when ignoreErrors=true. Set it with
+// WithLoadHint so that commands (push, diff) can provide context-appropriate
+// remediation advice without baking command-specific text into this package.
+type loadHintKey struct{}
+
+// WithLoadHint returns a copy of ctx that carries an additional hint message.
+// The message is logged as a second Warn line when manifest.Load encounters
+// errors and ignoreErrors=true. Pull --force does NOT set a hint so the user
+// is not told to run pull --force while they are already doing so.
+func WithLoadHint(ctx context.Context, hint string) context.Context {
+	return context.WithValue(ctx, loadHintKey{}, hint)
+}
+
 type InvalidManifestError struct {
 	error
 }
@@ -116,8 +130,27 @@ func Load(ctx context.Context, logger log.Logger, fs filesystem.Fs, envs env.Pro
 	m.repositories = content.Templates.Repositories
 
 	// Set records
-	if err := m.SetRecords(content.records()); err != nil && !ignoreErrors {
-		return nil, InvalidManifestError{errors.PrefixError(err, "invalid manifest")}
+	if err := m.SetRecords(content.records()); err != nil {
+		if !ignoreErrors {
+			return nil, InvalidManifestError{errors.PrefixError(err, "invalid manifest")}
+		}
+		// Log a warning so the user knows some records were skipped.
+		// This happens when a config's parent config is missing from the manifest
+		// (e.g. a scheduler config whose orchestrator was never pulled).
+		// Orphaned records are deleted by SetRecords, so no record is left with
+		// an unresolved parent path.
+		logger.Warn(ctx, "Manifest loaded with warnings (some records were skipped):")
+		var multi errors.MultiError
+		if errors.As(err, &multi) {
+			for _, e := range multi.WrappedErrors() {
+				logger.Warnf(ctx, "  - %s", e)
+			}
+		} else {
+			logger.Warnf(ctx, "  - %s", err)
+		}
+		if hint, _ := ctx.Value(loadHintKey{}).(string); hint != "" {
+			logger.Warn(ctx, hint)
+		}
 	}
 
 	// Return

internal/pkg/project/manifest/manifest_test.go

@@ -181,7 +181,12 @@ func TestManifestCyclicDependency(t *testing.T) {
 	manifest, err := Load(ctx, log.NewNopLogger(), fs, env.Empty(), false)
 	assert.Nil(t, manifest)
 	require.Error(t, err)
-	assert.Equal(t, "invalid manifest:\n- a cyclic relation was found when resolving path to config \"branch:123/component:keboola.variables/config:111\"", err.Error())
+	// Config 111 is removed due to the cyclic dependency. Config 222 (whose parent
+	// is config 111) is then also detected as having a missing parent and removed.
+	assert.Equal(t, "invalid manifest:"+
+		"\n- a cyclic relation was found when resolving path to config \"branch:123/component:keboola.variables/config:111\""+
+		"\n- manifest record for config \"branch:123/component:keboola.variables/config:111\" not found, referenced from config \"branch:123/component:keboola.variables/config:222\"",
+		err.Error())
 }
 
 func TestManifest_AllowTargetENV(t *testing.T) {

internal/pkg/service/cli/cmd/sync/diff/cmd.go

@@ -3,6 +3,7 @@ package diff
 import (
 	"github.com/spf13/cobra"
 
+	projectManifest "github.com/keboola/keboola-as-code/internal/pkg/project/manifest"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/cli/dependencies"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/cli/helpmsg"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/common/configmap"
@@ -43,8 +44,14 @@ func Command(p dependencies.Provider) *cobra.Command {
 				return err
 			}
 
-			// Get local project
-			prj, _, err := d.LocalProject(cmd.Context(), false)
+			// Get local project.
+			// Use ignoreErrors=true so that an inconsistent manifest (e.g. a scheduler
+			// whose orchestrator parent was never pulled) does not block diff.
+			// SetRecords() deletes any orphaned records, so no record is left with an
+			// unresolved parent path. A warning is logged by manifest.Load() in this case.
+			ctx := projectManifest.WithLoadHint(cmd.Context(),
+				"Run `kbc push` to clean up the manifest, or `kbc pull --force` to reset local state.")
+			prj, _, err := d.LocalProject(ctx, true)
 			if err != nil {
 				return err
 			}

internal/pkg/service/cli/cmd/sync/push/cmd.go

@@ -3,9 +3,11 @@ package push
 import (
 	"github.com/spf13/cobra"
 
+	projectManifest "github.com/keboola/keboola-as-code/internal/pkg/project/manifest"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/cli/dependencies"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/cli/helpmsg"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/common/configmap"
+	saveManifest "github.com/keboola/keboola-as-code/pkg/lib/operation/project/local/manifest/save"
 	"github.com/keboola/keboola-as-code/pkg/lib/operation/project/sync/push"
 	loadState "github.com/keboola/keboola-as-code/pkg/lib/operation/state/load"
 )
@@ -45,18 +47,47 @@ func Command(p dependencies.Provider) *cobra.Command {
 				return err
 			}
 
-			// Get local project
-			prj, _, err := d.LocalProject(cmd.Context(), false)
+			// Get local project.
+			// Use ignoreErrors=true so that an inconsistent manifest (e.g. a scheduler
+			// whose orchestrator parent was never pulled) does not block push.
+			// SetRecords() deletes any orphaned records, so no record is left with an
+			// unresolved parent path. A warning is logged by manifest.Load() in this case.
+			// NOTE: if the orphaned configs still exist in remote, running kbc push --force
+			// will schedule them for remote deletion (they are absent from local state).
+			hint := "Orphaned records are excluded from the push. Running `kbc push --force` may delete them from remote. " +
+				"Run `kbc pull --force` to reset local state."
+			if f.Force.Value {
+				hint = "Orphaned records are excluded from the push. " +
+					"With --force, these configs will be deleted from remote if they still exist there. " +
+					"Run `kbc pull --force` to reset local state first."
+			}
+			ctx := projectManifest.WithLoadHint(cmd.Context(), hint)
+			prj, _, err := d.LocalProject(ctx, true)
 			if err != nil {
 				return err
 			}
 
+			// Snapshot IsChanged() before LoadState to capture only orphan-cleanup changes.
+			// LoadState and push.Run do not modify the manifest (push is remote-only),
+			// so any post-snapshot IsChanged() would only be from orphan cleanup.
+			orphansCleaned := prj.ProjectManifest().IsChanged()
+
 			// Load project state
 			projectState, err := prj.LoadState(cmd.Context(), loadState.PushOptions(), d)
 			if err != nil {
 				return err
 			}
 
+			// If orphaned records were dropped during manifest load, the in-memory state
+			// diverges from the manifest file on disk. Save the cleaned manifest now so
+			// the warning does not recur on every subsequent push/diff invocation.
+			// Skip when --dry-run: that flag promises no local side effects.
+			if orphansCleaned && !f.DryRun.Value {
+				if _, err = saveManifest.Run(cmd.Context(), projectState.ProjectManifest(), projectState.Fs(), d); err != nil {
+					return err
+				}
+			}
+
 			// Change description - optional arg
 			changeDescription := "Updated from #KeboolaCLI"
 			if len(args) > 0 {

internal/pkg/state/manifest/records.go

@@ -34,35 +34,73 @@ func NewRecords(sortBy string) *Records {
 
 func (r *Records) SetRecords(records []model.ObjectManifest) error {
 	r.loaded = false
+	orphaned := false
 	defer func() {
-		// Track if manifest was changed after load
+		// Track if manifest was changed after load.
+		// If records were dropped (orphaned), the in-memory state diverges from
+		// the manifest file on disk — preserve changed=true so callers (e.g. push)
+		// can detect this and save the cleaned manifest to silence future warnings.
 		r.loaded = true
-		r.changed = false
+		if !orphaned {
+			r.changed = false
+		}
 	}()
 
 	// Clear
 	r.all = orderedmap.New()
 
-	// Track records
+	// Track records (only adds to r.all; parent-path resolution happens below).
+	// Do not return early on error: successfully tracked records must still have
+	// their parent paths resolved, otherwise they are left with parentPathSet=false
+	// which causes panics during subsequent state loading.
 	errs := errors.NewMultiError()
 	for _, record := range records {
 		if err := r.trackRecord(record); err != nil {
 			errs.Append(err)
+			orphaned = true
 		}
 	}
-	if errs.Len() > 0 {
-		return errs
+
+	// Resolve parent paths — snapshot keys first: orderedmap.Keys() returns the live slice, mutated by DeleteByKey.
+	// Child-before-parent ordering is safe: if a parent's PersistRecord fails, the cascade loop removes its children.
+	for _, key := range append([]string{}, r.all.Keys()...) {
+		v, found := r.all.Get(key)
+		if !found {
+			continue
+		}
+		record := v.(model.ObjectManifest)
+		if err := r.PersistRecord(record); err != nil {
+			errs.Append(err)
+			r.DeleteByKey(record.Key())
+			orphaned = true
+			r.changed = true // explicit: DeleteByKey only propagates changed for persisted records
+		}
 	}
 
-	// Resolve parent paths, we can do it when all the records are loaded
-	for _, key := range r.all.Keys() {
-		record, _ := r.all.Get(key)
-		if err := r.PersistRecord(record.(model.ObjectManifest)); err != nil {
-			return err
+	// Cascade: re-run until stable, detaching orphaned children whose parent was just removed.
+	for {
+		removed := false
+		for _, key := range append([]string{}, r.all.Keys()...) {
+			v, found := r.all.Get(key)
+			if !found {
+				continue
+			}
+			record := v.(model.ObjectManifest)
+			if _, err := r.GetParent(record); err != nil {
+				// Intentionally not appended to errs: this is a cascade effect of a
+				// parent already removed above. The root cause is already in errs.
+				r.naming.Detach(record.Key()) // must detach: PersistRecord already attached it
+				r.DeleteByKey(record.Key())   // r.changed implicit: records are persisted, unlike the PersistRecord failure path above
+				removed = true
+				orphaned = true
+			}
+		}
+		if !removed {
+			break
 		}
 	}
 
-	return nil
+	return errs.ErrorOrNil()
 }
 
 func (r *Records) SortBy() string {

internal/pkg/state/manifest/records_test.go

@@ -45,3 +45,81 @@ func TestManifestRecordGetParentNil(t *testing.T) {
 	assert.Nil(t, parent)
 	require.NoError(t, err)
 }
+
+// TestSetRecords_OrphanedScheduler_SiblingGetsPathResolved is a regression test
+// for the panic in AddRelatedPath that occurred when a manifest contained both an
+// orphaned scheduler and sibling configs that sorted after it.
+//
+// The pre-fix code did an early return from SetRecords on the first PersistRecord
+// failure. Any record that had not yet been processed was left with
+// parentPathSet=false. When a subsequent pull tried to write those records' files,
+// Path() returned only the bare relative path (without the branch prefix), so the
+// filesystem.IsFrom check in AddRelatedPath panicked.
+//
+// The fix continues processing all records and deletes only the failing ones, so
+// every surviving record has its parent path resolved before SetRecords returns.
+func TestSetRecords_OrphanedScheduler_SiblingGetsPathResolved(t *testing.T) {
+	t.Parallel()
+	r := NewRecords(model.SortByID)
+
+	branchManifest := &model.BranchManifest{
+		BranchKey: model.BranchKey{ID: 1},
+		// Simulate JSON-deserialized state: only RelativePath is set,
+		// parentPath and parentPathSet are zero values.
+		Paths: model.Paths{AbsPath: model.AbsPath{RelativePath: "main"}},
+	}
+	// Scheduler whose target orchestrator is absent from the manifest.
+	// Placed before the extractor so that the pre-fix early-return would leave
+	// the extractor unprocessed (parentPathSet=false).
+	schedulerManifest := &model.ConfigManifest{
+		ConfigKey: model.ConfigKey{
+			BranchID:    1,
+			ComponentID: "keboola.scheduler",
+			ID:          "456",
+		},
+		Paths: model.Paths{AbsPath: model.AbsPath{RelativePath: "schedules/scheduler"}},
+		Relations: model.Relations{
+			&model.SchedulerForRelation{
+				ComponentID: "keboola.orchestrator",
+				ConfigID:    "999", // deliberately absent from the manifest
+			},
+		},
+	}
+	// Sibling extractor with no special relations — its parent is the branch.
+	// After SetRecords it must have IsParentPathSet()==true and the full path
+	// "main/extractor/ex-generic-v2/empty" (branch prefix included).
+	// With the pre-fix code this record would have parentPathSet=false, causing
+	// a panic in AddRelatedPath when the pull wrote meta.json under the config.
+	extractorManifest := &model.ConfigManifest{
+		ConfigKey: model.ConfigKey{
+			BranchID:    1,
+			ComponentID: "ex-generic-v2",
+			ID:          "789",
+		},
+		Paths: model.Paths{AbsPath: model.AbsPath{RelativePath: "extractor/ex-generic-v2/empty"}},
+	}
+
+	err := r.SetRecords([]model.ObjectManifest{branchManifest, schedulerManifest, extractorManifest})
+
+	// The orphaned scheduler is reported but does not hard-fail the whole load.
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), `manifest record for config "branch:1/component:keboola.orchestrator/config:999" not found`)
+
+	// Orphaned scheduler must be deleted.
+	_, found := r.GetRecord(schedulerManifest.Key())
+	assert.False(t, found, "orphaned scheduler must be removed from records")
+
+	// Branch must survive and be resolved.
+	branchRecord, found := r.GetRecord(branchManifest.Key())
+	require.True(t, found, "branch record must remain")
+	assert.True(t, branchRecord.IsParentPathSet())
+
+	// Sibling extractor must survive with its parent path correctly resolved so
+	// that Path() returns "main/extractor/ex-generic-v2/empty" and not the bare
+	// relative path "extractor/ex-generic-v2/empty" (which would panic later).
+	extRecord, found := r.GetRecord(extractorManifest.Key())
+	require.True(t, found, "sibling extractor must not be removed")
+	assert.True(t, extRecord.IsParentPathSet(),
+		"sibling extractor must have parent path resolved even after the orphaned scheduler fails")
+	assert.Equal(t, "main/extractor/ex-generic-v2/empty", extRecord.Path())
+}

test/cli/create/config-data-gateway/out/main/application/keboola.app-data-gateway/new-config/config.json

@@ -1,5 +1,20 @@
 {
   "parameters": {
-    "action": ""
+    "db": {
+      "#privateKey": "",
+      "account": "",
+      "database": "",
+      "driver": "",
+      "host": "",
+      "loginType": "",
+      "port": 443,
+      "region": "",
+      "role": "",
+      "schema": "",
+      "ssoLoginAvailable": false,
+      "user": "",
+      "warehouse": "",
+      "workspaceId": 0
+    }
   }
 }

test/cli/create/config/out/main/extractor/keboola.ex-db-mysql/new-config/config.json

@@ -10,7 +10,6 @@
       "networkCompression": false,
       "port": 0,
       "ssh": {
-        "#privateKey": "",
         "enabled": false,
         "keys": {
           "#private": "",

test/cli/diff/diff-with-orphaned-scheduler/args

@@ -0,0 +1 @@
+diff --storage-api-token %%TEST_KBC_STORAGE_API_TOKEN%%

test/cli/diff/diff-with-orphaned-scheduler/expected-code

@@ -0,0 +1 @@
+0