From d2a7af744defb74ddae2755cc1fa65805edb80e6 Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Thu, 18 Dec 2025 16:32:26 +1000 Subject: [PATCH] fix: uncloud daemon crash when listing partially replicated container records --- internal/machine/store/container.go | 41 ++++++++++++++++++++++---- internal/machine/store/store.go | 45 +++++++++++++++++++++++++---- pkg/api/container.go | 4 +++ 3 files changed, 78 insertions(+), 12 deletions(-) diff --git a/internal/machine/store/container.go b/internal/machine/store/container.go index 281804ec..a3772051 100644 --- a/internal/machine/store/container.go +++ b/internal/machine/store/container.go @@ -82,7 +82,7 @@ func (s *Store) CreateOrUpdateContainer(ctx context.Context, ctr api.ServiceCont // ListContainers returns a list of container records from the store database that match the given options. func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]ContainerRecord, error) { - q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers"). + q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers"). Where(sq.Eq{"sync_status": SyncStatusSynced}) if len(opts.MachineIDs) > 0 { @@ -112,14 +112,23 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain defer rows.Close() var containers []ContainerRecord - var cJSON, machineID, syncStatus, updatedAtStr string + var id, cJSON, machineID, syncStatus, updatedAtStr string var updatedAt time.Time + skipped := 0 for rows.Next() { - if err = rows.Scan(&cJSON, &machineID, &syncStatus, &updatedAtStr); err != nil { + if err = rows.Scan(&id, &cJSON, &machineID, &syncStatus, &updatedAtStr); err != nil { return nil, fmt.Errorf("scan container record: %w", err) } + // Skip containers with empty JSON data. This can happen during partial replication + // when cr-sqlite has created the row but the container column hasn't been synced yet. + if cJSON == "" || cJSON == "{}" { + slog.Debug("Skipping container with empty data in the store (partial replication?).", "id", id) + skipped++ + continue + } + var c api.ServiceContainer if err = json.Unmarshal([]byte(cJSON), &c); err != nil { return nil, fmt.Errorf("unmarshal container: %w", err) @@ -135,6 +144,11 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain }) } + if skipped > 0 { + slog.Warn("Listing containers from the store skipped empty records (possibly due to partial replication).", + "skipped", skipped, "valid", len(containers)) + } + return containers, nil } @@ -166,7 +180,7 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error // receive any values, it just signals when a container(s) has been added, updated, or deleted in the database. func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-chan struct{}, error) { // TODO: figure out whether we need sync_status at all (not used at the moment). - q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers"). + q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers"). Where(sq.Eq{"sync_status": SyncStatusSynced}) query, args, err := q.ToSql() if err != nil { @@ -179,15 +193,24 @@ func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-c } var containers []ContainerRecord - var cJSON, updatedAtStr string + var id, cJSON, updatedAtStr string + skipped := 0 rows := sub.Rows() for rows.Next() { var cr ContainerRecord - if err = rows.Scan(&cJSON, &cr.MachineID, &cr.SyncStatus, &updatedAtStr); err != nil { + if err = rows.Scan(&id, &cJSON, &cr.MachineID, &cr.SyncStatus, &updatedAtStr); err != nil { return nil, nil, err } + // Skip containers with empty JSON data. This can happen during partial replication + // when cr-sqlite has created the row but the container column hasn't been synced yet. + if cJSON == "" || cJSON == "{}" { + slog.Debug("Skipping container with empty data in the store (partial replication?).", "id", id) + skipped++ + continue + } + if err = json.Unmarshal([]byte(cJSON), &cr.Container); err != nil { return nil, nil, fmt.Errorf("unmarshal container: %w", err) } @@ -196,6 +219,12 @@ func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-c } containers = append(containers, cr) } + + if skipped > 0 { + slog.Warn("Container subscription skipped empty records in the store (possibly due to partial replication).", + "skipped", skipped, "valid", len(containers)) + } + events, err := sub.Changes() if err != nil { return nil, nil, fmt.Errorf("get subscription changes: %w", err) diff --git a/internal/machine/store/store.go b/internal/machine/store/store.go index 58d44e05..10096681 100644 --- a/internal/machine/store/store.go +++ b/internal/machine/store/store.go @@ -116,19 +116,29 @@ func (s *Store) GetMachine(ctx context.Context, machineID string) (*pb.MachineIn } func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) { - rows, err := s.corro.QueryContext(ctx, "SELECT info FROM machines ORDER BY name") + rows, err := s.corro.QueryContext(ctx, "SELECT id, info FROM machines ORDER BY name") if err != nil { return nil, err } defer rows.Close() var machines []*pb.MachineInfo + skipped := 0 + for rows.Next() { - var mJSON string - if err = rows.Scan(&mJSON); err != nil { + var id, mJSON string + if err = rows.Scan(&id, &mJSON); err != nil { return nil, err } + // Skip machines with empty JSON data. This can happen during partial replication + // when cr-sqlite has created the row but the info column hasn't been synced yet. + if mJSON == "" || mJSON == "{}" { + slog.Debug("Skipping machine with empty data in the store (partial replication?).", "id", id) + skipped++ + continue + } + protojsonParser := protojson.UnmarshalOptions{DiscardUnknown: true} var m pb.MachineInfo if err = protojsonParser.Unmarshal([]byte(mJSON), &m); err != nil { @@ -141,6 +151,12 @@ func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) { } machines = append(machines, &m) } + + if skipped > 0 { + slog.Warn("Listing machines from the store skipped empty records (possibly due to partial replication).", + "skipped", skipped, "valid", len(machines)) + } + return machines, nil } @@ -186,24 +202,41 @@ func (s *Store) DeleteMachine(ctx context.Context, id string) error { // SubscribeMachines returns a list of machines and a channel that signals changes to the list. The channel doesn't // receive any values, it just signals when a machine has been added, updated, or deleted in the database. func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-chan struct{}, error) { - sub, err := s.corro.SubscribeContext(ctx, "SELECT info FROM machines ORDER BY name", nil, false) + sub, err := s.corro.SubscribeContext(ctx, "SELECT id, info FROM machines ORDER BY name", nil, false) if err != nil { return nil, nil, err } rows := sub.Rows() var machines []*pb.MachineInfo + skipped := 0 + for rows.Next() { - var mJSON string - if err = rows.Scan(&mJSON); err != nil { + var id, mJSON string + if err = rows.Scan(&id, &mJSON); err != nil { return nil, nil, err } + + // Skip machines with empty JSON data. This can happen during partial replication + // when cr-sqlite has created the row but the info column hasn't been synced yet. + if mJSON == "" || mJSON == "{}" { + slog.Debug("Skipping machine with empty data in the store (partial replication?).", "id", id) + skipped++ + continue + } + var m pb.MachineInfo if err = protojson.Unmarshal([]byte(mJSON), &m); err != nil { return nil, nil, fmt.Errorf("unmarshal machine info: %w", err) } machines = append(machines, &m) } + + if skipped > 0 { + slog.Warn("Machine subscription skipped empty records in the store (possibly due to partial replication).", + "skipped", skipped, "valid", len(machines)) + } + events, err := sub.Changes() if err != nil { return nil, nil, fmt.Errorf("get subscription changes: %w", err) diff --git a/pkg/api/container.go b/pkg/api/container.go index cab49443..e386cfe3 100644 --- a/pkg/api/container.go +++ b/pkg/api/container.go @@ -135,6 +135,10 @@ func (c *Container) UnmarshalJSON(data []byte) error { } *c = Container(temp) + if c.ContainerJSONBase == nil { + return fmt.Errorf("container data is missing mandatory base fields: %s", data) + } + c.Name = strings.TrimPrefix(c.Name, "/") return nil