Skip to main content

internal/infra/pg/rebuildjob.go

internal/infra/pg · 312 lines · 23 declarations · source

Declarations​

var ErrRebuildBusy​

var ErrRebuildBusy = errors.New("a project rebuild is already active")

source

var ErrRebuildFenced​

var ErrRebuildFenced = errors.New("project rebuild runner no longer owns the operation")

source

var ErrRebuildNotFound​

var ErrRebuildNotFound = errors.New("project rebuild not found")

source

var ErrRebuildPoolCapacity​

var ErrRebuildPoolCapacity = errors.New("project rebuild requires a database pool with at least two connections")

source

type RebuildJob​

type RebuildJob struct {
	ID              string    `json:"job_id"`
	Version         string    `json:"extractor_version"`
	Through         int64     `json:"through_offset"`
	After           int64     `json:"after_offset"`
	Pending         *int64    `json:"pending_offset,omitempty"`
	Rebuilt         int64     `json:"sources_rebuilt"`
	Skipped         int64     `json:"sources_skipped"`
	Status          string    `json:"status"`
	CancelRequested bool      `json:"cancel_requested"`
	CreatedAt       time.Time `json:"created_at"`
	UpdatedAt       time.Time `json:"updated_at"`
	lease           string
}

source

type RebuildJobStore​

type RebuildJobStore struct{ pool *pgxpool.Pool }

source

func NewRebuildJobStore​

func NewRebuildJobStore(pool *pgxpool.Pool) *RebuildJobStore

source

type RebuildJobSession​

type RebuildJobSession struct {
	store                   *RebuildJobStore
	conn                    *pgxpool.Conn
	schema                  Schema
	scope, key, lease, lock string
}

The connection holds only a session lock while the model runs, never an open transaction. A replacement runner changes the persisted lease so a disconnected predecessor cannot publish.

source

type GenerationJobGuard​

type GenerationJobGuard struct {
	JobID, LeaseID string
	Offset         int64
}

source

type RebuildWork​

type RebuildWork struct {
	SourceID, Key   string
	Offset          int64
	CancelRequested bool
}

source

const readRebuildJobSQL​

const readRebuildJobSQL = `SELECT job_id::text,extractor_version,through_offset,after_offset,pending_offset,
    sources_rebuilt,sources_skipped,status,cancel_requested,created_at,updated_at,lease_id::text
    FROM {schema}.fact_rebuild_job WHERE scope=$1 AND job_id=$2::uuid`

source

func scanRebuildJob​

func scanRebuildJob(row pgx.Row) (RebuildJob, error)

source

func validRebuildJob​

func validRebuildJob(scope, key string) error

source

method RebuildJobStore.Status​

func (s *RebuildJobStore) Status(ctx context.Context, schema Schema, scope, key string) (RebuildJob, error)

source

method RebuildJobStore.Acquire​

func (s *RebuildJobStore) Acquire(ctx context.Context, schema Schema, scope, key, version, actor string) (*RebuildJobSession, error)

source

method RebuildJobSession.Close​

func (s *RebuildJobSession) Close()

source

method RebuildJobSession.Guard​

func (s *RebuildJobSession) Guard(offset int64) *GenerationJobGuard

source

func sourceRebuildKey​

func sourceRebuildKey(job string, offset int64) string

source

method RebuildJobSession.lockedJob​

func (s *RebuildJobSession) lockedJob(ctx context.Context, tx pgx.Tx) (RebuildJob, error)

source

method RebuildJobSession.Next​

func (s *RebuildJobSession) Next(ctx context.Context) (RebuildWork, bool, error)

Reserving an offset is the cancellation boundary. A pending source may finish; cancellation prevents reserving a later one. The offset survives erasure without retaining an observation ID.

source

method RebuildJobSession.Complete​

func (s *RebuildJobSession) Complete(ctx context.Context, offset int64, rebuilt bool) error

source

method RebuildJobStore.Cancel​

func (s *RebuildJobStore) Cancel(ctx context.Context, schema Schema, scope, key, actor string) (RebuildJob, error)

source

func validateGenerationJobTx​

func validateGenerationJobTx(ctx context.Context, tx pgx.Tx, schema Schema, snapshot GenerationSnapshot, key, version string, g *GenerationJobGuard) error

Take the job row before source locks. Cancel only changes the job row; erasure never needs it.

source