featurebase/dax/models/job.go
Travis Turner 8fca15e936
RetryWithTx (#2348)
* First pass at RetryWithTx

* Refactor RetryWithTx to take a writable bool (instead of reads, writes)

* Implment DirectiveMethodDiff

This commit adds support for a Directive to contain only the diffs (as
opposed to the full Directive).

* Update controller tests to allow for DirectiveMethodDiff (over Full)

* Update RetryWithTx to retry on duplicate key constraint.

If two concurrent processes call IngestShard() for the same shard, both
were trying to insert the same job into the jobs table. That resulted in
a duplicate key error from the database. We want to include that error
in the list of errors for which RetryWithTx should retry.

* Remove unused method: Directive.TranslatePartitions()

* Replace query in a loop with a single query

We had a query which was looking to see if a job already existed. That
query was inside a loop, and could potentially generate 256 queries (for
example). This commit replaces that logic so that we use a single query
wiht an `IN ()` clause.

* Convert to directive version-by-address

This commit uses a separate directive version per address. It moves the
version get/increment back inside the buildDirective method so that if
two concurrent processes are building a directive for the same address,
one of them will get rolled back trying to commit the version update.

* Migration for directive version by address

* Add a comment about DirectiveVersion lock/unlock logic

* Remove AddLastWins

* fix linter

* handle error in walkdir

* fix test failures from removing AddLastWins
2023-03-30 20:54:37 -05:00

70 lines
2.2 KiB
Go

package models
import (
"encoding/json"
"time"
"github.com/featurebasedb/featurebase/v3/dax"
"github.com/gobuffalo/nulls"
"github.com/gobuffalo/pop/v6"
"github.com/gobuffalo/validate/v3"
"github.com/gobuffalo/validate/v3/validators"
"github.com/gofrs/uuid"
)
// Job represents a job which can be assigned to a worker or free (unassigned).
type Job struct {
ID uuid.UUID `json:"id" db:"id"`
Name dax.Job `json:"name" db:"name"`
Role dax.RoleType `json:"role" db:"role"`
DatabaseID dax.DatabaseID `json:"database_id" db:"database_id"`
WorkerID nulls.UUID `json:"-" db:"worker_id"`
Worker *Worker `json:"worker" db:"-" belongs_to:"worker"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
}
// String is not required by pop and may be deleted
func (t *Job) String() string {
jt, _ := json.MarshalIndent(t, " ", " ") //nolint:errchkjson
return string(jt)
}
// Jobs is not required by pop and may be deleted
type Jobs []Job
// String is not required by pop and may be deleted
func (t Jobs) String() string {
jt, _ := json.MarshalIndent(t, " ", " ") //nolint:errchkjson
return string(jt)
}
// Contains returns true if j is in Jobs.
func (t Jobs) Contains(j dax.Job) bool {
for i := range t {
if t[i].Name == j {
return true
}
}
return false
}
// Validate gets run every time you call a "pop.Validate*" (pop.ValidateAndSave, pop.ValidateAndCreate, pop.ValidateAndUpdate) method.
// This method is not required and may be deleted.
func (t *Job) Validate(tx *pop.Connection) (*validate.Errors, error) {
return validate.Validate(
&validators.StringIsPresent{Field: string(t.Name), Name: "Name"},
), nil
}
// ValidateCreate gets run every time you call "pop.ValidateAndCreate" method.
// This method is not required and may be deleted.
func (t *Job) ValidateCreate(tx *pop.Connection) (*validate.Errors, error) {
return validate.NewErrors(), nil
}
// ValidateUpdate gets run every time you call "pop.ValidateAndUpdate" method.
// This method is not required and may be deleted.
func (t *Job) ValidateUpdate(tx *pop.Connection) (*validate.Errors, error) {
return validate.NewErrors(), nil
}