Propagate error for table migration

pkg/controller/chi/worker-migrator.go

@@ -105,6 +105,7 @@ func (w *worker) migrateTables(ctx context.Context, host *api.Host, opts *migrat
 			M(host).F().
 			Error("ERROR add tables failed on shard/host:%d/%d cluster:%s err:%v",
 				host.Runtime.Address.ShardIndex, host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ClusterName, err)
+		return err
 	}
 
 	w.a.V(1).

pkg/controller/chi/worker-reconciler-chi.go

@@ -857,9 +857,11 @@ func (w *worker) reconcileHost(ctx context.Context, host *api.Host) error {
 		return err
 	}
 	if err := w.reconcileHostMain(ctx, host); err != nil {
+		// Domain/schema failure: keep status as Requested (excluded from remote_servers)
+		// and do not include the host into cluster activities.
 		return err
 	}
-	// Host is now added and functional
+	// Host STS is up and schema migration (if required) succeeded — safe to include.
 
 	if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) {
 		host.GetReconcileAttributes().SetStatus(types.ObjectStatusCreated)
@@ -984,18 +986,18 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error {
 			Warning("Reconcile Host Main - unable to reconcile PVC. Host: %s Err: %v", host.GetName(), err)
 	}
 
-	// Finalize main reconcile with domain activities
+	// Finalize main reconcile with domain activities (schema migration, FIPS checks).
+	// Schema/domain failures must fail this host's reconcile so we do not:
+	//   1. flip status Requested → Created (which re-includes the host in remote_servers),
+	//   2. call includeHost / mark the host ready for Distributed traffic,
+	// while local tables are still missing. Callers of reconcileHost only include
+	// the host after reconcileHostMain returns nil.
 	if err := w.reconcileHostMainDomain(ctx, host, migrateTableOpts); err != nil {
+		metrics.HostReconcilesErrors(ctx, host.GetCR())
 		w.a.V(1).
 			M(host).F().
 			Warning("Reconcile Host Main - unable to reconcile domain reconcile. Host: %s Err: %v", host.GetName(), err)
-		// Propagate abort signals (e.g. runtime FIPS image-policy violation) so
-		// reconcile() routes them through markReconcileCompletedUnsuccessfully,
-		// which persists StatusAborted via updateCRObjectStatus. Other errors
-		// stay logged but non-fatal — preserves pre-existing tolerance.
-		if errors.Is(err, common.ErrCRUDAbort) {
-			return err
-		}
+		return err
 	}
 
 	return nil

pkg/model/chi/schemer/cluster.go

@@ -53,20 +53,24 @@ func (c *Cluster) SetClusterConnectionParams(clusterConnectionParams *clickhouse
 func (c *Cluster) queryUnzipColumns(ctx context.Context, hosts []string, sql string, columns ...*[]string) error {
 	if util.IsContextDone(ctx) {
 		log.V(1).Info("ctx is done")
-		return nil
+		return ctx.Err()
 	}
 
 	if len(hosts) == 0 {
 		// Nowhere to fetch data from
 		return nil
 	}
 
-	// Fetch data from any of specified hosts
+	// Fetch data from any of specified hosts. Propagate failures so schema migration
+	// cannot silently treat "could not discover objects" as "nothing to create".
 	query, err := c.SetHosts(hosts).QueryAny(ctx, sql)
 	if err != nil {
-		return nil
+		return err
 	}
 	if query == nil {
+		if util.IsContextDone(ctx) {
+			return ctx.Err()
+		}
 		return nil
 	}
 

pkg/model/chi/schemer/distributed.go

@@ -56,27 +56,36 @@ func (s *ClusterSchemer) getDistributedObjectsSQLs(ctx context.Context, host *ap
 	// Exclude self — see the matching comment in getReplicatedObjectsSQLs.
 	peers := s.Names(interfaces.NameFQDNs, host, api.Cluster{}, true)
 
-	databaseNames, createDatabaseSQLs := debugCreateSQLs(
-		s.QueryUnzip2Columns(
-			ctx,
-			peers,
-			s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName),
-		),
+	databaseNames, createDatabaseSQLs, err := s.QueryUnzip2Columns(
+		ctx,
+		peers,
+		s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName),
 	)
-	tableNames, createTableSQLs := debugCreateSQLs(
-		s.QueryUnzipAndApplyUUIDs(
-			ctx,
-			peers,
-			s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName),
-		),
+	databaseNames, createDatabaseSQLs = debugCreateSQLs(databaseNames, createDatabaseSQLs, err)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	tableNames, createTableSQLs, err := s.QueryUnzipAndApplyUUIDs(
+		ctx,
+		peers,
+		s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName),
 	)
-	functionNames, createFunctionSQLs := debugCreateSQLs(
-		s.QueryUnzip2Columns(
-			ctx,
-			peers,
-			s.sqlCreateFunction(host.Runtime.Address.ClusterName),
-		),
+	tableNames, createTableSQLs = debugCreateSQLs(tableNames, createTableSQLs, err)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	functionNames, createFunctionSQLs, err := s.QueryUnzip2Columns(
+		ctx,
+		peers,
+		s.sqlCreateFunction(host.Runtime.Address.ClusterName),
 	)
+	functionNames, createFunctionSQLs = debugCreateSQLs(functionNames, createFunctionSQLs, err)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	return util.ConcatSlices([][]string{databaseNames, tableNames, functionNames}),
 		util.ConcatSlices([][]string{createDatabaseSQLs, createTableSQLs, createFunctionSQLs}),
 		nil

pkg/model/chi/schemer/replicated.go

@@ -71,27 +71,36 @@ func (s *ClusterSchemer) getReplicatedObjectsSQLs(ctx context.Context, host *api
 	// schemer silently skips CREATE. Peers hold the truth.
 	peers := s.Names(interfaces.NameFQDNs, host, api.Cluster{}, true)
 
-	databaseNames, createDatabaseSQLs := debugCreateSQLs(
-		s.QueryUnzip2Columns(
-			ctx,
-			peers,
-			s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName),
-		),
+	databaseNames, createDatabaseSQLs, err := s.QueryUnzip2Columns(
+		ctx,
+		peers,
+		s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName),
 	)
-	tableNames, createTableSQLs := debugCreateSQLs(
-		s.QueryUnzipAndApplyUUIDs(
-			ctx,
-			peers,
-			s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName),
-		),
+	databaseNames, createDatabaseSQLs = debugCreateSQLs(databaseNames, createDatabaseSQLs, err)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	tableNames, createTableSQLs, err := s.QueryUnzipAndApplyUUIDs(
+		ctx,
+		peers,
+		s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName),
 	)
-	functionNames, createFunctionSQLs := debugCreateSQLs(
-		s.QueryUnzip2Columns(
-			ctx,
-			peers,
-			s.sqlCreateFunction(host.Runtime.Address.ClusterName),
-		),
+	tableNames, createTableSQLs = debugCreateSQLs(tableNames, createTableSQLs, err)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	functionNames, createFunctionSQLs, err := s.QueryUnzip2Columns(
+		ctx,
+		peers,
+		s.sqlCreateFunction(host.Runtime.Address.ClusterName),
 	)
+	functionNames, createFunctionSQLs = debugCreateSQLs(functionNames, createFunctionSQLs, err)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	return util.ConcatSlices([][]string{databaseNames, tableNames, functionNames}),
 		util.ConcatSlices([][]string{createDatabaseSQLs, createTableSQLs, createFunctionSQLs}),
 		nil

pkg/model/chi/schemer/schemer.go

@@ -86,23 +86,24 @@ func (s *ClusterSchemer) createTablesSQLs(
 	replicatedCreateSQLs []string,
 	distributedObjectNames []string,
 	distributedCreateSQLs []string,
+	err error,
 ) {
-	if names, sql, err := s.getReplicatedObjectsSQLs(ctx, host); err == nil {
-		replicatedObjectNames = names
-		replicatedCreateSQLs = sql
+	replicatedObjectNames, replicatedCreateSQLs, err = s.getReplicatedObjectsSQLs(ctx, host)
+	if err != nil {
+		return nil, nil, nil, nil, err
 	}
-	if names, sql, err := s.getDistributedObjectsSQLs(ctx, host); err == nil {
-		distributedObjectNames = names
-		distributedCreateSQLs = sql
+	distributedObjectNames, distributedCreateSQLs, err = s.getDistributedObjectsSQLs(ctx, host)
+	if err != nil {
+		return nil, nil, nil, nil, err
 	}
-	return
+	return replicatedObjectNames, replicatedCreateSQLs, distributedObjectNames, distributedCreateSQLs, nil
 }
 
 // HostCreateTables creates tables on a new host
 func (s *ClusterSchemer) HostCreateTables(ctx context.Context, host *api.Host) error {
 	if util.IsContextDone(ctx) {
 		log.V(1).Info("ctx is done")
-		return nil
+		return ctx.Err()
 	}
 
 	log.V(1).M(host).F().S().Info("Migrating schema objects to host %s", host.Runtime.Address.HostName)
@@ -111,29 +112,29 @@ func (s *ClusterSchemer) HostCreateTables(ctx context.Context, host *api.Host) e
 	replicatedObjectNames,
 		replicatedCreateSQLs,
 		distributedObjectNames,
-		distributedCreateSQLs := s.createTablesSQLs(ctx, host)
+		distributedCreateSQLs,
+		err := s.createTablesSQLs(ctx, host)
+	if err != nil {
+		log.V(1).M(host).F().Error("Failed to discover schema objects for host %s: %v", host.Runtime.Address.HostName, err)
+		return err
+	}
 
-	var err1 error
 	if len(replicatedCreateSQLs) > 0 {
 		log.V(1).M(host).F().Info("Creating replicated objects at %s: %v", host.Runtime.Address.HostName, replicatedObjectNames)
 		log.V(2).M(host).F().Info("\n%v", replicatedCreateSQLs)
-		err1 = s.ExecHost(ctx, host, replicatedCreateSQLs,
-			clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true))
+		if err := s.ExecHost(ctx, host, replicatedCreateSQLs,
+			clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)); err != nil {
+			return err
+		}
 	}
 
-	var err2 error
 	if len(distributedCreateSQLs) > 0 {
 		log.V(1).M(host).F().Info("Creating distributed objects at %s: %v", host.Runtime.Address.HostName, distributedObjectNames)
 		log.V(2).M(host).F().Info("\n%v", distributedCreateSQLs)
-		err2 = s.ExecHost(ctx, host, distributedCreateSQLs,
-			clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true))
-	}
-
-	if err2 != nil {
-		return err2
-	}
-	if err1 != nil {
-		return err1
+		if err := s.ExecHost(ctx, host, distributedCreateSQLs,
+			clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)); err != nil {
+			return err
+		}
 	}
 
 	return nil