Compare commits
1 Commits
7f37bda8b8
...
ea552ec448
Author | SHA1 | Date | |
---|---|---|---|
|
ea552ec448 |
@ -6,10 +6,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||||||
|
|
||||||
## [Unreleased]
|
## [Unreleased]
|
||||||
|
|
||||||
## [0.1.0] - 2025-01-19
|
## [0.1.0] - 2025-01-18
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
- add dead letter queue
|
|
||||||
- move schedules to registered workers
|
- move schedules to registered workers
|
||||||
- prune old workers
|
- prune old workers
|
||||||
- deregister worker on close
|
- deregister worker on close
|
||||||
|
@ -3,7 +3,6 @@ package app
|
|||||||
import (
|
import (
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/deadletter"
|
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/executor"
|
"git.front.kjuulh.io/kjuulh/orbis/internal/executor"
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/modelschedule"
|
"git.front.kjuulh.io/kjuulh/orbis/internal/modelschedule"
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/scheduler"
|
"git.front.kjuulh.io/kjuulh/orbis/internal/scheduler"
|
||||||
@ -49,11 +48,7 @@ func (a *App) WorkScheduler() *workscheduler.WorkScheduler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (a *App) WorkProcessor() *workprocessor.WorkProcessor {
|
func (a *App) WorkProcessor() *workprocessor.WorkProcessor {
|
||||||
return workprocessor.NewWorkProcessor(a.WorkScheduler(), a.logger, a.DeadLetter())
|
return workprocessor.NewWorkProcessor(a.WorkScheduler(), a.logger)
|
||||||
}
|
|
||||||
|
|
||||||
func (a *App) DeadLetter() *deadletter.DeadLetter {
|
|
||||||
return deadletter.NewDeadLetter(Postgres(), a.logger)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *App) ModelSchedule() *modelschedule.ModelSchedule {
|
func (a *App) ModelSchedule() *modelschedule.ModelSchedule {
|
||||||
|
@ -1,35 +0,0 @@
|
|||||||
package deadletter
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"log/slog"
|
|
||||||
|
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/deadletter/repositories"
|
|
||||||
"github.com/google/uuid"
|
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
|
||||||
)
|
|
||||||
|
|
||||||
type DeadLetter struct {
|
|
||||||
db *pgxpool.Pool
|
|
||||||
logger *slog.Logger
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewDeadLetter(db *pgxpool.Pool, logger *slog.Logger) *DeadLetter {
|
|
||||||
return &DeadLetter{
|
|
||||||
db: db,
|
|
||||||
logger: logger,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (d *DeadLetter) InsertDeadLetter(ctx context.Context, schedule uuid.UUID) error {
|
|
||||||
repo := repositories.New(d.db)
|
|
||||||
|
|
||||||
d.logger.WarnContext(ctx, "deadlettering schedule", "schedule", schedule)
|
|
||||||
if err := repo.InsertDeadLetter(ctx, schedule); err != nil {
|
|
||||||
return fmt.Errorf("failed to insert item into dead letter: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
|
|
||||||
}
|
|
@ -1,12 +0,0 @@
|
|||||||
-- name: Ping :one
|
|
||||||
SELECT 1;
|
|
||||||
|
|
||||||
-- name: InsertDeadLetter :exec
|
|
||||||
INSERT INTO dead_letter
|
|
||||||
(
|
|
||||||
schedule_id
|
|
||||||
)
|
|
||||||
VALUES
|
|
||||||
(
|
|
||||||
$1
|
|
||||||
);
|
|
@ -1,32 +0,0 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
|
||||||
// versions:
|
|
||||||
// sqlc v1.23.0
|
|
||||||
|
|
||||||
package repositories
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
|
|
||||||
"github.com/jackc/pgx/v5"
|
|
||||||
"github.com/jackc/pgx/v5/pgconn"
|
|
||||||
)
|
|
||||||
|
|
||||||
type DBTX interface {
|
|
||||||
Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
|
|
||||||
Query(context.Context, string, ...interface{}) (pgx.Rows, error)
|
|
||||||
QueryRow(context.Context, string, ...interface{}) pgx.Row
|
|
||||||
}
|
|
||||||
|
|
||||||
func New(db DBTX) *Queries {
|
|
||||||
return &Queries{db: db}
|
|
||||||
}
|
|
||||||
|
|
||||||
type Queries struct {
|
|
||||||
db DBTX
|
|
||||||
}
|
|
||||||
|
|
||||||
func (q *Queries) WithTx(tx pgx.Tx) *Queries {
|
|
||||||
return &Queries{
|
|
||||||
db: tx,
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,35 +0,0 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
|
||||||
// versions:
|
|
||||||
// sqlc v1.23.0
|
|
||||||
|
|
||||||
package repositories
|
|
||||||
|
|
||||||
import (
|
|
||||||
"github.com/google/uuid"
|
|
||||||
"github.com/jackc/pgx/v5/pgtype"
|
|
||||||
)
|
|
||||||
|
|
||||||
type DeadLetter struct {
|
|
||||||
ScheduleID uuid.UUID `json:"schedule_id"`
|
|
||||||
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type ModelSchedule struct {
|
|
||||||
ModelName string `json:"model_name"`
|
|
||||||
LastRun pgtype.Timestamptz `json:"last_run"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type WorkSchedule struct {
|
|
||||||
ScheduleID uuid.UUID `json:"schedule_id"`
|
|
||||||
WorkerID uuid.UUID `json:"worker_id"`
|
|
||||||
StartRun pgtype.Timestamptz `json:"start_run"`
|
|
||||||
EndRun pgtype.Timestamptz `json:"end_run"`
|
|
||||||
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
||||||
State string `json:"state"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type WorkerRegister struct {
|
|
||||||
WorkerID uuid.UUID `json:"worker_id"`
|
|
||||||
Capacity int32 `json:"capacity"`
|
|
||||||
HeartBeat pgtype.Timestamptz `json:"heart_beat"`
|
|
||||||
}
|
|
@ -1,18 +0,0 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
|
||||||
// versions:
|
|
||||||
// sqlc v1.23.0
|
|
||||||
|
|
||||||
package repositories
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
|
|
||||||
"github.com/google/uuid"
|
|
||||||
)
|
|
||||||
|
|
||||||
type Querier interface {
|
|
||||||
InsertDeadLetter(ctx context.Context, scheduleID uuid.UUID) error
|
|
||||||
Ping(ctx context.Context) (int32, error)
|
|
||||||
}
|
|
||||||
|
|
||||||
var _ Querier = (*Queries)(nil)
|
|
@ -1,39 +0,0 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
|
||||||
// versions:
|
|
||||||
// sqlc v1.23.0
|
|
||||||
// source: queries.sql
|
|
||||||
|
|
||||||
package repositories
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
|
|
||||||
"github.com/google/uuid"
|
|
||||||
)
|
|
||||||
|
|
||||||
const insertDeadLetter = `-- name: InsertDeadLetter :exec
|
|
||||||
INSERT INTO dead_letter
|
|
||||||
(
|
|
||||||
schedule_id
|
|
||||||
)
|
|
||||||
VALUES
|
|
||||||
(
|
|
||||||
$1
|
|
||||||
)
|
|
||||||
`
|
|
||||||
|
|
||||||
func (q *Queries) InsertDeadLetter(ctx context.Context, scheduleID uuid.UUID) error {
|
|
||||||
_, err := q.db.Exec(ctx, insertDeadLetter, scheduleID)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
const ping = `-- name: Ping :one
|
|
||||||
SELECT 1
|
|
||||||
`
|
|
||||||
|
|
||||||
func (q *Queries) Ping(ctx context.Context) (int32, error) {
|
|
||||||
row := q.db.QueryRow(ctx, ping)
|
|
||||||
var column_1 int32
|
|
||||||
err := row.Scan(&column_1)
|
|
||||||
return column_1, err
|
|
||||||
}
|
|
@ -1,21 +0,0 @@
|
|||||||
version: "2"
|
|
||||||
sql:
|
|
||||||
- queries: queries.sql
|
|
||||||
schema: ../persistence/migrations/
|
|
||||||
engine: "postgresql"
|
|
||||||
gen:
|
|
||||||
go:
|
|
||||||
out: "repositories"
|
|
||||||
package: "repositories"
|
|
||||||
sql_package: "pgx/v5"
|
|
||||||
emit_json_tags: true
|
|
||||||
emit_prepared_queries: true
|
|
||||||
emit_interface: true
|
|
||||||
emit_empty_slices: true
|
|
||||||
emit_result_struct_pointers: true
|
|
||||||
emit_params_struct_pointers: true
|
|
||||||
overrides:
|
|
||||||
- db_type: "uuid"
|
|
||||||
go_type:
|
|
||||||
import: "github.com/google/uuid"
|
|
||||||
type: "UUID"
|
|
@ -9,11 +9,6 @@ import (
|
|||||||
"github.com/jackc/pgx/v5/pgtype"
|
"github.com/jackc/pgx/v5/pgtype"
|
||||||
)
|
)
|
||||||
|
|
||||||
type DeadLetter struct {
|
|
||||||
ScheduleID uuid.UUID `json:"schedule_id"`
|
|
||||||
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type ModelSchedule struct {
|
type ModelSchedule struct {
|
||||||
ModelName string `json:"model_name"`
|
ModelName string `json:"model_name"`
|
||||||
LastRun pgtype.Timestamptz `json:"last_run"`
|
LastRun pgtype.Timestamptz `json:"last_run"`
|
||||||
|
@ -20,8 +20,3 @@ CREATE TABLE work_schedule (
|
|||||||
|
|
||||||
CREATE INDEX idx_work_schedule_worker ON work_schedule (worker_id);
|
CREATE INDEX idx_work_schedule_worker ON work_schedule (worker_id);
|
||||||
CREATE INDEX idx_work_schedule_worker_updated ON work_schedule (worker_id, updated_at DESC);
|
CREATE INDEX idx_work_schedule_worker_updated ON work_schedule (worker_id, updated_at DESC);
|
||||||
|
|
||||||
CREATE TABLE dead_letter (
|
|
||||||
schedule_id UUID PRIMARY KEY NOT NULL
|
|
||||||
, updated_at TIMESTAMPTZ NOT NULL default now()
|
|
||||||
);
|
|
||||||
|
@ -9,11 +9,6 @@ import (
|
|||||||
"github.com/jackc/pgx/v5/pgtype"
|
"github.com/jackc/pgx/v5/pgtype"
|
||||||
)
|
)
|
||||||
|
|
||||||
type DeadLetter struct {
|
|
||||||
ScheduleID uuid.UUID `json:"schedule_id"`
|
|
||||||
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type ModelSchedule struct {
|
type ModelSchedule struct {
|
||||||
ModelName string `json:"model_name"`
|
ModelName string `json:"model_name"`
|
||||||
LastRun pgtype.Timestamptz `json:"last_run"`
|
LastRun pgtype.Timestamptz `json:"last_run"`
|
||||||
|
@ -4,10 +4,8 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"math/rand/v2"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/deadletter"
|
|
||||||
"git.front.kjuulh.io/kjuulh/orbis/internal/workscheduler"
|
"git.front.kjuulh.io/kjuulh/orbis/internal/workscheduler"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
)
|
)
|
||||||
@ -15,18 +13,12 @@ import (
|
|||||||
type WorkProcessor struct {
|
type WorkProcessor struct {
|
||||||
workscheduler *workscheduler.WorkScheduler
|
workscheduler *workscheduler.WorkScheduler
|
||||||
logger *slog.Logger
|
logger *slog.Logger
|
||||||
deadletter *deadletter.DeadLetter
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewWorkProcessor(
|
func NewWorkProcessor(workscheduler *workscheduler.WorkScheduler, logger *slog.Logger) *WorkProcessor {
|
||||||
workscheduler *workscheduler.WorkScheduler,
|
|
||||||
logger *slog.Logger,
|
|
||||||
deadletter *deadletter.DeadLetter,
|
|
||||||
) *WorkProcessor {
|
|
||||||
return &WorkProcessor{
|
return &WorkProcessor{
|
||||||
workscheduler: workscheduler,
|
workscheduler: workscheduler,
|
||||||
logger: logger,
|
logger: logger,
|
||||||
deadletter: deadletter,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -49,17 +41,10 @@ func (w *WorkProcessor) ProcessNext(ctx context.Context, workerID uuid.UUID) err
|
|||||||
}
|
}
|
||||||
|
|
||||||
time.Sleep(time.Millisecond * 10)
|
time.Sleep(time.Millisecond * 10)
|
||||||
// Process or achive
|
|
||||||
|
|
||||||
if err := w.workscheduler.Archive(ctx, *schedule); err != nil {
|
if err := w.workscheduler.Archive(ctx, *schedule); err != nil {
|
||||||
return fmt.Errorf("failed to archive item: %w", err)
|
return fmt.Errorf("failed to archive item: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if rand.Float64() > 0.9 {
|
|
||||||
if err := w.deadletter.InsertDeadLetter(ctx, *schedule); err != nil {
|
|
||||||
return fmt.Errorf("failed to deadletter message: %w", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
@ -9,11 +9,6 @@ import (
|
|||||||
"github.com/jackc/pgx/v5/pgtype"
|
"github.com/jackc/pgx/v5/pgtype"
|
||||||
)
|
)
|
||||||
|
|
||||||
type DeadLetter struct {
|
|
||||||
ScheduleID uuid.UUID `json:"schedule_id"`
|
|
||||||
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type ModelSchedule struct {
|
type ModelSchedule struct {
|
||||||
ModelName string `json:"model_name"`
|
ModelName string `json:"model_name"`
|
||||||
LastRun pgtype.Timestamptz `json:"last_run"`
|
LastRun pgtype.Timestamptz `json:"last_run"`
|
||||||
|
Loading…
x
Reference in New Issue
Block a user