New registers the replica with the database and periodically updates to ensure it's healthy. It contacts all other alive replicas to ensure they are reachable.
(ctx context.Context, logger slog.Logger, db database.Store, ps pubsub.Pubsub, options *Options)
| 42 | // ensure it's healthy. It contacts all other alive replicas to ensure they are |
| 43 | // reachable. |
| 44 | func New(ctx context.Context, logger slog.Logger, db database.Store, ps pubsub.Pubsub, options *Options) (*Manager, error) { |
| 45 | if options == nil { |
| 46 | options = &Options{} |
| 47 | } |
| 48 | if options.ID == uuid.Nil { |
| 49 | options.ID = uuid.New() |
| 50 | } |
| 51 | if options.PeerTimeout == 0 { |
| 52 | options.PeerTimeout = 3 * time.Second |
| 53 | } |
| 54 | if options.UpdateInterval == 0 { |
| 55 | options.UpdateInterval = 5 * time.Second |
| 56 | } |
| 57 | if options.CleanupInterval == 0 { |
| 58 | // The cleanup interval can be quite long, because it's |
| 59 | // primary purpose is to clean up dead replicas. |
| 60 | options.CleanupInterval = 30 * time.Minute |
| 61 | } |
| 62 | hostname := cliutil.Hostname() |
| 63 | databaseLatency, err := db.Ping(ctx) |
| 64 | if err != nil { |
| 65 | return nil, xerrors.Errorf("ping database: %w", err) |
| 66 | } |
| 67 | // nolint:gocritic // Inserting a replica is a system function. |
| 68 | replica, err := db.InsertReplica(dbauthz.AsSystemRestricted(ctx), database.InsertReplicaParams{ |
| 69 | ID: options.ID, |
| 70 | CreatedAt: dbtime.Now(), |
| 71 | StartedAt: dbtime.Now(), |
| 72 | UpdatedAt: dbtime.Now(), |
| 73 | Hostname: hostname, |
| 74 | RegionID: options.RegionID, |
| 75 | RelayAddress: options.RelayAddress, |
| 76 | Version: buildinfo.Version(), |
| 77 | // #nosec G115 - Safe conversion for microseconds latency which is expected to be within int32 range |
| 78 | DatabaseLatency: int32(databaseLatency.Microseconds()), |
| 79 | Primary: true, |
| 80 | }) |
| 81 | if err != nil { |
| 82 | return nil, xerrors.Errorf("insert replica: %w", err) |
| 83 | } |
| 84 | err = ps.Publish(PubsubEvent, []byte(options.ID.String())) |
| 85 | if err != nil { |
| 86 | return nil, xerrors.Errorf("publish new replica: %w", err) |
| 87 | } |
| 88 | ctx, cancelFunc := context.WithCancel(ctx) |
| 89 | manager := &Manager{ |
| 90 | id: options.ID, |
| 91 | options: options, |
| 92 | db: db, |
| 93 | pubsub: ps, |
| 94 | self: replica, |
| 95 | logger: logger, |
| 96 | closed: make(chan struct{}), |
| 97 | closeCancel: cancelFunc, |
| 98 | } |
| 99 | err = manager.syncReplicas(ctx) |
| 100 | if err != nil { |
| 101 | return nil, xerrors.Errorf("run replica: %w", err) |