git.giacomolli.uk - onehouse/blob - driver/ledger.go

   1 package driver
   2
   3 import (
   4         "fmt"
   5         "sync"
   6
   7         "onehouse/protocol"
   8 )
   9
  10 const commandResultCapacity = 256
  11
  12 // Ledger keeps the first result of each of the most recent 256 commands for one
  13 // device, in insertion order. Its methods are safe for concurrent use. A single
  14 // queue consumer checks Lookup before acting and calls Remember before sending
  15 // the result, also after a failed send. The ledger does not reserve physical
  16 // work, persist records, or prevent replay after eviction.
  17 type Ledger struct {
  18         deviceID string
  19         mu       sync.Mutex
  20         results  map[string]ResultRecord
  21         order    []string
  22 }
  23
  24 // NewLedger creates an empty result ledger for one canonical device ID.
  25 func NewLedger(deviceID string) (*Ledger, error) {
  26         if !protocol.ValidID(deviceID) {
  27                 return nil, fmt.Errorf("ledger device ID must be a canonical ULID")
  28         }
  29
  30         return &Ledger{
  31                 deviceID: deviceID,
  32                 results:  make(map[string]ResultRecord),
  33                 order:    make([]string, 0, commandResultCapacity),
  34         }, nil
  35 }
  36
  37 // Lookup returns an independent copy of a result without changing its age.
  38 func (l *Ledger) Lookup(commandID string) (ResultRecord, bool) {
  39         l.mu.Lock()
  40         defer l.mu.Unlock()
  41
  42         result, exists := l.results[commandID]
  43
  44         return result.clone(), exists
  45 }
  46
  47 // Remember stores the first validated result before publication. An existing
  48 // result wins even when a duplicate proposes another status or reason. Its ID
  49 // and observation time remain stable after failed publication and on replay.
  50 func (l *Ledger) Remember(commandID string, payload protocol.ResultPayload) (ResultRecord, error) {
  51         l.mu.Lock()
  52         defer l.mu.Unlock()
  53
  54         result, exists := l.results[commandID]
  55         if exists {
  56                 return result.clone(), nil
  57         }
  58
  59         metadata, err := newMetadata(protocol.Metadata{CorrelationID: commandID})
  60         if err != nil {
  61                 return ResultRecord{}, fmt.Errorf("create result metadata: %w", err)
  62         }
  63
  64         _, err = protocol.BuildResult(l.deviceID, commandID, metadata, payload)
  65         if err != nil {
  66                 return ResultRecord{}, fmt.Errorf("validate result: %w", err)
  67         }
  68
  69         if len(l.order) == commandResultCapacity {
  70                 delete(l.results, l.order[0])
  71
  72                 l.order = l.order[1:]
  73         }
  74
  75         result = ResultRecord{
  76                 Metadata: metadata,
  77                 Payload:  payload,
  78         }
  79         l.results[commandID] = result
  80         l.order = append(l.order, commandID)
  81
  82         return result.clone(), nil
  83 }