Documentation
¶
Overview ¶
Package store persists the message log and the PNR state.
The two are deliberately separate concerns with different guarantees.
The message log is append-only and holds the exact bytes received or sent. It is written before anything is interpreted, so a parser bug costs a reprocessing run rather than a lost booking, and so "what did we actually receive at 14:32" has an answer that does not depend on the parser that was deployed at the time.
PNR state is derived. Every change is recorded as an event carrying the id of the message that caused it, and the current record is a projection of those events. That is what makes an interline dispute answerable: the record shows what it holds now, and the events show which partner message put it there.
Index ¶
- Constants
- Variables
- func MigrateSchema(ctx context.Context, s *Postgres) error
- func Migrations() ([]migration, error)
- func NormaliseFlightKey(key string) string
- func SchemaSQL() (string, error)
- func SegmentOnFlight(seg *pnr.Segment, flightKey, wireDate string) bool
- type Diagnostic
- type Direction
- type Event
- type Format
- type Lookup
- type Mem
- func (s *Mem) AppendMessage(ctx context.Context, m *Message) error
- func (s *Mem) Close() error
- func (s *Mem) CreatePNR(ctx context.Context, p *pnr.PNR, events []Event) error
- func (s *Mem) DividePNR(ctx context.Context, parent *pnr.PNR, expected int64, child *pnr.PNR, ...) error
- func (s *Mem) Enqueue(ctx context.Context, item *QueueItem) error
- func (s *Mem) Events(ctx context.Context, pnrID string) ([]Event, error)
- func (s *Mem) FindByDedupKey(ctx context.Context, peer, key string) (string, bool, error)
- func (s *Mem) FindOutboundByKey(ctx context.Context, peer, key string) (string, bool, error)
- func (s *Mem) FindPNRByDocument(ctx context.Context, compactNumber string) (*pnr.PNR, error)
- func (s *Mem) FindPNRByExternalLocator(ctx context.Context, owner, value string) (*pnr.PNR, error)
- func (s *Mem) FindPNRsByFlight(ctx context.Context, flightKey, wireDate string, limit int) ([]*pnr.PNR, error)
- func (s *Mem) FindPNRsDueBy(ctx context.Context, deadline time.Time, limit int) ([]*pnr.PNR, error)
- func (s *Mem) FindPNRsStale(ctx context.Context, before time.Time, limit int) ([]*pnr.PNR, error)
- func (s *Mem) GetMessage(ctx context.Context, id string) (*Message, error)
- func (s *Mem) GetPNR(ctx context.Context, locator string) (*pnr.PNR, error)
- func (s *Mem) GetPNRByID(ctx context.Context, id string) (*pnr.PNR, error)
- func (s *Mem) ListMessages(ctx context.Context, f MessageFilter) ([]*Message, error)
- func (s *Mem) ListPNRs(ctx context.Context, limit int) ([]*pnr.PNR, error)
- func (s *Mem) ListQueue(ctx context.Context, f QueueFilter) ([]*QueueItem, error)
- func (s *Mem) NextLocatorCounter(ctx context.Context) (uint64, error)
- func (s *Mem) QueueCounts(ctx context.Context) (map[string]int, error)
- func (s *Mem) UpdateMessage(ctx context.Context, m *Message) error
- func (s *Mem) UpdatePNR(ctx context.Context, p *pnr.PNR, expected int64, events []Event) error
- func (s *Mem) WorkQueueItem(ctx context.Context, id, by, note string) error
- type Message
- type MessageFilter
- type Postgres
- func (s *Postgres) AppendMessage(ctx context.Context, m *Message) error
- func (s *Postgres) Close() error
- func (s *Postgres) CreatePNR(ctx context.Context, p *pnr.PNR, events []Event) error
- func (s *Postgres) DividePNR(ctx context.Context, parent *pnr.PNR, expected int64, child *pnr.PNR, ...) error
- func (s *Postgres) Enqueue(ctx context.Context, item *QueueItem) error
- func (s *Postgres) Events(ctx context.Context, pnrID string) ([]Event, error)
- func (s *Postgres) FindByDedupKey(ctx context.Context, peer, key string) (string, bool, error)
- func (s *Postgres) FindOutboundByKey(ctx context.Context, peer, key string) (string, bool, error)
- func (s *Postgres) FindPNRByDocument(ctx context.Context, compactNumber string) (*pnr.PNR, error)
- func (s *Postgres) FindPNRByExternalLocator(ctx context.Context, owner, value string) (*pnr.PNR, error)
- func (s *Postgres) FindPNRsByFlight(ctx context.Context, flightKey, wireDate string, limit int) ([]*pnr.PNR, error)
- func (s *Postgres) FindPNRsDueBy(ctx context.Context, deadline time.Time, limit int) ([]*pnr.PNR, error)
- func (s *Postgres) FindPNRsStale(ctx context.Context, before time.Time, limit int) ([]*pnr.PNR, error)
- func (s *Postgres) GetMessage(ctx context.Context, id string) (*Message, error)
- func (s *Postgres) GetPNR(ctx context.Context, locator string) (*pnr.PNR, error)
- func (s *Postgres) GetPNRByID(ctx context.Context, id string) (*pnr.PNR, error)
- func (s *Postgres) ListMessages(ctx context.Context, f MessageFilter) ([]*Message, error)
- func (s *Postgres) ListPNRs(ctx context.Context, limit int) ([]*pnr.PNR, error)
- func (s *Postgres) ListQueue(ctx context.Context, f QueueFilter) ([]*QueueItem, error)
- func (s *Postgres) NextLocatorCounter(ctx context.Context) (uint64, error)
- func (s *Postgres) QueueCounts(ctx context.Context) (map[string]int, error)
- func (s *Postgres) UpdateMessage(ctx context.Context, m *Message) error
- func (s *Postgres) UpdatePNR(ctx context.Context, p *pnr.PNR, expected int64, events []Event) error
- func (s *Postgres) WorkQueueItem(ctx context.Context, id, by, note string) error
- type QueueFilter
- type QueueItem
- type QueueStore
- type Status
- type Store
Constants ¶
const ( // QueueConfirmation holds records a partner has confirmed, where the // holding side has to be told or the itinerary reissued. QueueConfirmation = "confirmation" // QueueUnable holds records a partner refused. Someone has to rebook. QueueUnable = "unable" // QueueWaitlist holds records with waitlisted segments awaiting clearance. QueueWaitlist = "waitlist" // QueuePending holds requests a partner has not answered inside the agreed // time. Nothing is wrong with the record; the conversation has stalled. QueuePending = "pending" // QueueTicketing holds records whose ticketing time limit is near or past. QueueTicketing = "ticketing" // QueueScheduleChange holds records touched by a schedule message: the // flight they are booked on has moved, changed equipment or been // cancelled, and the passenger has not been told. QueueScheduleChange = "schedule-change" // QueueDivergence holds records where our state and a partner's disagree, // which is the case a human must always see. QueueDivergence = "divergence" // QueueGeneral is the fallback for placements with no better home. QueueGeneral = "general" )
A queue is a list of records waiting for someone to do something about them.
It is the mechanism a reservations system uses to turn an asynchronous partner conversation into work: a carrier confirms a segment overnight, a ticketing deadline passes, a request goes unanswered, and each of those has to surface somewhere a person or a robot will look. Without queues those events are visible only to whoever happens to re-read the record.
Queue names here are Jetway's own vocabulary, not an IATA standard. Real systems number their queues and the numbering is house-specific, so a deployment that has to match an existing convention should map these names onto its own numbers at the edge rather than expect the numbers to be meaningful across systems.
Variables ¶
var ( // ErrNotFound is returned when a record or message does not exist. ErrNotFound = errors.New("store: not found") // ErrConflict is returned when an update's expected version does not match // the stored version. The caller must re-read and retry: a blind overwrite // loses whichever concurrent change it did not see. ErrConflict = errors.New("store: version conflict") // ErrDuplicate is returned when a record locator is already taken. ErrDuplicate = errors.New("store: duplicate") )
Errors returned by every implementation.
var Queues = []string{ QueueConfirmation, QueueUnable, QueueWaitlist, QueuePending, QueueTicketing, QueueScheduleChange, QueueDivergence, QueueGeneral, }
Queues lists the known queue names in the order a console should show them.
Functions ¶
func MigrateSchema ¶
MigrateSchema applies any migrations the database has not seen.
Each migration runs inside its own transaction together with the row that records it, so a failure part-way leaves the database at a version that actually reflects its contents. Applying is idempotent, which is what makes running it on every start safe and removes a manual deployment step.
func Migrations ¶
func Migrations() ([]migration, error)
Migrations returns the embedded schema changes in version order.
func NormaliseFlightKey ¶
NormaliseFlightKey renders a carrier and flight number in the one spelling the lookups compare against.
Carriers write the same flight both zero-padded and bare, sometimes in the same conversation, so a key is only useful if both spellings collapse onto it. The designator is taken as the first two characters, never by scanning for the first digit: IATA designators are two characters and a third of them are alphanumeric -- U2, 4U, 2B -- so "the number starts at the first digit" split easyJet's own designator in half and no U2 flight could ever be matched to a schedule change. Found by the world simulator, whose carriers come from the real registry; every hand-written test here had used BA.
func SchemaSQL ¶
SchemaSQL returns every migration concatenated, for inspection and for bootstrapping a database by hand.
func SegmentOnFlight ¶
SegmentOnFlight reports whether a segment is a live holding on a flight.
Both backends decide with this function rather than each with its own predicate. The Postgres lookup narrows with an index and then asks this, so an index that over-matches -- and containment does, because it will match a cancelled segment as happily as a live one -- cannot change the answer.
Types ¶
type Diagnostic ¶
type Diagnostic struct {
Layer string `json:"layer"` // "typeb", "edifact", "airimp", "padis"
Severity string `json:"severity"`
Code string `json:"code"`
Detail string `json:"detail"`
Line int `json:"line,omitempty"`
}
Diagnostic is a decoder observation, flattened across codec packages so the console and the API present one shape.
type Direction ¶
type Direction string
Direction distinguishes traffic we received from traffic we sent.
type Event ¶
type Event struct {
ID string `json:"id"`
PNRID string `json:"pnr_id"`
Seq int64 `json:"seq"`
Type string `json:"type"`
Detail string `json:"detail"`
Payload json.RawMessage `json:"payload,omitempty"`
MessageID string `json:"message_id,omitempty"`
Actor string `json:"actor,omitempty"`
At time.Time `json:"at"`
}
Event is one applied change to a record.
type Lookup ¶
type Lookup interface {
// FindPNRByDocument returns the record holding a document, by its compact
// thirteen-digit number. Not found is (nil, nil): a document this node
// never issued is an ordinary answer, not a failure.
FindPNRByDocument(ctx context.Context, compactNumber string) (*pnr.PNR, error)
// FindPNRByExternalLocator returns the record carrying another system's
// locator. Owner may be empty to match any.
FindPNRByExternalLocator(ctx context.Context, owner, value string) (*pnr.PNR, error)
// FindPNRsByFlight returns every live record holding a segment on a
// flight. wireDate may be empty to match every date, which is what a
// schedule message covering a period needs.
//
// flightKey is the carrier and flight number with leading zeros removed,
// because carriers write the same flight both ways and a schedule change
// that misses half its holdings is worse than useless.
FindPNRsByFlight(ctx context.Context, flightKey, wireDate string, limit int) ([]*pnr.PNR, error)
// FindPNRsStale returns live records untouched since before the given
// time, most overdue first.
//
// The ordering is the point. The sweeper used to read the most recently
// updated records and look for stale ones among them, which is inverted:
// the freshest records are precisely not the stale ones. Ordering by the
// thing that makes a record due means a limit drops the least urgent work
// rather than all of it.
FindPNRsStale(ctx context.Context, before time.Time, limit int) ([]*pnr.PNR, error)
// FindPNRsDueBy returns live records owing a ticketing time limit before
// the given time, soonest deadline first.
FindPNRsDueBy(ctx context.Context, deadline time.Time, limit int) ([]*pnr.PNR, error)
}
Lookups that find a record by something other than its own locator.
These exist because the alternative was a bug rather than merely a slow query. Three places on the hot path used to walk ListPNRs, which is "ORDER BY updated_at DESC LIMIT n" -- so they searched only recently touched records and reported *not found* for anything older. A ticket control message about a booking made last month was refused with "no record holds this document", which is not slow, it is false, and it gets more likely as the store grows.
So the contract here is total: an implementation must search every record or return an error. It must never quietly answer from a prefix.
type Mem ¶
type Mem struct {
// MaxMessages and MaxRecords bound what is retained, oldest discarded
// first. Zero means unbounded, which is right for a test and wrong for
// anything reachable from the internet: without a bound, a public demo is
// a memory leak with a submit button.
MaxMessages int
MaxRecords int
// Now, when set, stamps defaults instead of the wall clock. Set before
// use; read without a lock.
Now func() time.Time
// contains filtered or unexported fields
}
Mem is an in-memory Store.
It exists so the gateway runs with no external dependency -- for the demo, for tests, and for a carrier evaluating the message flow before provisioning a database. It is not a production backend: nothing survives a restart.
func (*Mem) FindByDedupKey ¶
func (*Mem) FindOutboundByKey ¶
func (*Mem) FindPNRByDocument ¶
func (*Mem) FindPNRByExternalLocator ¶
func (*Mem) FindPNRsByFlight ¶
func (*Mem) FindPNRsDueBy ¶
func (*Mem) FindPNRsStale ¶
func (*Mem) ListMessages ¶
func (*Mem) NextLocatorCounter ¶
type Message ¶
type Message struct {
ID string `json:"id"`
Direction Direction `json:"direction"`
At time.Time `json:"at"`
// Transport and Peer say how it arrived and from whom. Peer is the link
// name, which is what routing and per-partner policy key off.
Transport string `json:"transport"`
Peer string `json:"peer"`
Format Format `json:"format"`
// Kind is the decoded message type, e.g. "AIRIMP/sell" or "PAORES".
Kind string `json:"kind,omitempty"`
// Raw is the exact bytes on the wire. Never regenerate it from a parse.
Raw []byte `json:"-"`
// SHA256 is the hex digest of Raw, used for content-based deduplication
// where an application-level reference is unavailable.
SHA256 string `json:"sha256"`
Size int `json:"size"`
Status Status `json:"status"`
Error string `json:"error,omitempty"`
// DedupKey is the application-level idempotency key: an EDIFACT interchange
// control reference, or a Type B origin and time group. Empty when the
// message class carries none.
DedupKey string `json:"dedup_key,omitempty"`
// TraceID and SpanID tie this message to the trace that handled it. They
// turn "what happened to this message" from a search into a link, and they
// survive in the log after the trace itself has been sampled away.
TraceID string `json:"trace_id,omitempty"`
SpanID string `json:"span_id,omitempty"`
// PossibleDuplicate records the Type B PDM indicator. Inbound it means the
// sender said this may be a retransmission; outbound it means we marked it
// as one on redelivery. A duplicate that arrives flagged is the protocol
// working; one that arrives unflagged is worth an operator's attention.
PossibleDuplicate bool `json:"possible_duplicate,omitempty"`
// PNRID links the message to the record it touched.
PNRID string `json:"pnr_id,omitempty"`
// CorrelationID ties a response back to the request that provoked it, and
// is what turns a message list into a conversation.
CorrelationID string `json:"correlation_id,omitempty"`
Diagnostics []Diagnostic `json:"diagnostics,omitempty"`
}
Message is one unit of traffic, inbound or outbound.
type MessageFilter ¶
type MessageFilter struct {
Peer string
PNRID string
Status Status
Limit int
// SinceID returns only messages with an id greater than this one, which is
// how the console tails the log without re-fetching.
SinceID string
}
MessageFilter narrows a message listing.
type Postgres ¶
type Postgres struct {
// Now, when set, stamps defaults instead of the wall clock. Set before
// use; read without a lock.
Now func() time.Time
// contains filtered or unexported fields
}
Postgres is the production Store.
func OpenPostgres ¶
OpenPostgres connects and verifies the schema is present.
func (*Postgres) AppendMessage ¶
func (*Postgres) FindByDedupKey ¶
func (*Postgres) FindOutboundByKey ¶
func (*Postgres) FindPNRByDocument ¶
func (*Postgres) FindPNRByExternalLocator ¶
func (*Postgres) FindPNRsByFlight ¶
func (*Postgres) FindPNRsDueBy ¶
func (*Postgres) FindPNRsStale ¶
func (*Postgres) GetMessage ¶
func (*Postgres) GetPNRByID ¶
func (*Postgres) ListMessages ¶
func (*Postgres) NextLocatorCounter ¶
func (*Postgres) QueueCounts ¶
func (*Postgres) UpdateMessage ¶
type QueueFilter ¶
type QueueFilter struct {
// Queue restricts to one queue name. Empty means every queue.
Queue string
// PNRID restricts to one record.
PNRID string
// IncludeWorked returns cleared items as well as pending ones. The default
// is the working view: what still needs doing.
IncludeWorked bool
Limit int
}
QueueFilter narrows a queue listing.
type QueueItem ¶
type QueueItem struct {
ID string `json:"id"`
Queue string `json:"queue"`
PNRID string `json:"pnr_id"`
Locator string `json:"locator,omitempty"`
// Code is the stable machine-readable reason, e.g. "tktl_expired". Together
// with Queue, PNRID and SegmentRef it is the idempotency key: the same
// segment is on a queue once per reason until the item is worked, so a
// sweeper may run as often as it likes.
Code string `json:"code"`
// Reason is the human-readable form of the same thing.
Reason string `json:"reason"`
// MessageID is the message that caused the placement, where there was one.
MessageID string `json:"message_id,omitempty"`
// SegmentRef narrows the placement to one segment, where it applies.
SegmentRef int `json:"segment_ref,omitempty"`
PlacedAt time.Time `json:"placed_at"`
PlacedBy string `json:"placed_by"`
// WorkedAt is nil while the item is pending.
WorkedAt *time.Time `json:"worked_at,omitempty"`
WorkedBy string `json:"worked_by,omitempty"`
// Note is what the worker said when clearing it.
Note string `json:"note,omitempty"`
}
QueueItem is one record placed on one queue for one reason.
An item is evidence as much as it is a task: it says what put the record here, which message caused it, and when. Working an item does not delete it, because "who cleared this and when" is exactly the question asked after an interline dispute.
type QueueStore ¶
type QueueStore interface {
// Enqueue places a record on a queue. It returns ErrDuplicate when the same
// segment of the same record is already pending on the same queue for the
// same code, which is what makes a repeatedly-running sweeper harmless.
Enqueue(ctx context.Context, item *QueueItem) error
// WorkQueueItem marks an item done. Working an already-worked item returns
// ErrConflict rather than silently overwriting who cleared it.
WorkQueueItem(ctx context.Context, id, by, note string) error
// ListQueue returns items newest first.
ListQueue(ctx context.Context, f QueueFilter) ([]*QueueItem, error)
// QueueCounts returns the number of pending items per queue name.
QueueCounts(ctx context.Context) (map[string]int, error)
}
QueueStore is the queue half of the persistence contract. It is separate from Store only for readability; every backend implements both.
type Status ¶
type Status string
Status tracks a message through the pipeline.
const ( // StatusReceived means the bytes are durable but nothing has read them. StatusReceived Status = "received" // StatusDecoded means the envelope and body parsed. StatusDecoded Status = "decoded" // StatusApplied means the message changed a record, or was correctly // determined to require no change. StatusApplied Status = "applied" // StatusRejected means the message was understood and refused, for example // as a duplicate or as test traffic on a production link. StatusRejected Status = "rejected" // StatusDLQ means the pipeline could not process it and a human must look. // Messages never leave the system on this path; they wait to be replayed. StatusDLQ Status = "dlq" // StatusSent applies to outbound traffic handed to a transport. StatusSent Status = "sent" // StatusUndeliverable applies to outbound traffic no transport accepted. StatusUndeliverable Status = "undeliverable" // StatusAcknowledged means a partner confirmed receipt of outbound traffic, // which on an EDIFACT link is a CONTRL saying so. Delivery and // acknowledgement are different facts: a transport that took the bytes // proves nothing about whether the partner could read them. StatusAcknowledged Status = "acknowledged" // StatusRefused means a partner received outbound traffic and rejected it. StatusRefused Status = "refused" )
type Store ¶
type Store interface {
// AppendMessage writes a message. The raw bytes must be durable before it
// returns, because the caller acknowledges the peer on success.
AppendMessage(ctx context.Context, m *Message) error
// UpdateMessage records a status transition and any diagnostics.
UpdateMessage(ctx context.Context, m *Message) error
GetMessage(ctx context.Context, id string) (*Message, error)
ListMessages(ctx context.Context, f MessageFilter) ([]*Message, error)
// FindByDedupKey returns the id of an earlier inbound message from the same
// peer with the same application-level key, which is how a retransmission
// is recognised without re-applying it.
FindByDedupKey(ctx context.Context, peer, key string) (string, bool, error)
// FindOutboundByKey returns the id of a message sent to a peer carrying the
// given application-level key. It is the other direction of the same
// question, and is how an acknowledgement is matched to what it
// acknowledges.
FindOutboundByKey(ctx context.Context, peer, key string) (string, bool, error)
// CreatePNR stores a new record at version 1 along with the events that
// created it.
CreatePNR(ctx context.Context, p *pnr.PNR, events []Event) error
// UpdatePNR stores a record, failing with ErrConflict unless the stored
// version still equals expectedVersion.
UpdatePNR(ctx context.Context, p *pnr.PNR, expectedVersion int64, events []Event) error
GetPNR(ctx context.Context, locator string) (*pnr.PNR, error)
GetPNRByID(ctx context.Context, id string) (*pnr.PNR, error)
ListPNRs(ctx context.Context, limit int) ([]*pnr.PNR, error)
Events(ctx context.Context, pnrID string) ([]Event, error)
// NextLocatorCounter returns a value that has never been returned before.
// It is the uniqueness source behind record locator allocation.
NextLocatorCounter(ctx context.Context) (uint64, error)
// Lookup finds records by document, external locator or flight.
// DividePNR writes a division: the parent as it now stands and the new
// child, atomically, under the parent's expected version.
//
// It is one method rather than a create followed by an update because a
// division is one change to two records. Done as two writes, a version
// conflict on the second leaves the child created and the parent not
// updated -- both records then list the same passengers, which is a torn
// booking that no partner can be told about coherently. Under concurrency
// that is not a rare case: a carrier reply landing between the read and
// the write is exactly what optimistic concurrency is there to catch.
DividePNR(ctx context.Context, parent *pnr.PNR, expected int64, child *pnr.PNR,
parentEvents, childEvents []Event) error
Lookup
// QueueStore holds the work queues records are placed on.
QueueStore
Close() error
}
Store is the persistence contract.
Implementations must be safe for concurrent use: a gateway processes many links at once, and two of them can touch the same record.