80 lines
2.5 KiB
Go
80 lines
2.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"regexp"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"mesh-delivery-forge-bridge/internal/bridge"
|
|
)
|
|
|
|
// inputsBucket is the module's state bucket inputs (manifest state entry inputs): what the bridge read of a module
|
|
// directory, its manifest and lock-file hashes, by the directory's git tree hash, so a merge reads the forge only for the
|
|
// directories it changed (issue 1126; internal/bridge kept.go). A tree hash names its bytes, so a value is never stale;
|
|
// a merge adds a value only for each directory it changed (a few kilobytes each).
|
|
const inputsBucket = forgeBridgeModule + "_inputs"
|
|
|
|
// keptTimeout bounds one read or write of the bucket: the bucket is a saving, never a reason to fail a merge.
|
|
const keptTimeout = 2 * time.Second
|
|
|
|
var treeHash = regexp.MustCompile(`^[0-9a-f]{40}$|^[0-9a-f]{64}$`)
|
|
|
|
// kept is the bridge's Kept: in memory always, and in the bucket inputs when it is on the bus, so the first merge after a
|
|
// start reads only what changed too. A value the bucket cannot give or take is read from the forge, and said.
|
|
type kept struct {
|
|
kv jetstream.KeyValue // nil when the bucket is not on the bus
|
|
logger *log.Logger
|
|
mu sync.Mutex
|
|
mem map[string]bridge.Inputs
|
|
}
|
|
|
|
func newKept(kv jetstream.KeyValue, logger *log.Logger) *kept {
|
|
return &kept{kv: kv, logger: logger, mem: map[string]bridge.Inputs{}}
|
|
}
|
|
|
|
func (k *kept) Get(tree string) (bridge.Inputs, bool) {
|
|
k.mu.Lock()
|
|
in, ok := k.mem[tree]
|
|
k.mu.Unlock()
|
|
if ok || k.kv == nil || !treeHash.MatchString(tree) {
|
|
return in, ok
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), keptTimeout)
|
|
defer cancel()
|
|
e, err := k.kv.Get(ctx, tree)
|
|
if err != nil {
|
|
return bridge.Inputs{}, false // not kept (or not readable): read from the forge
|
|
}
|
|
if err := json.Unmarshal(e.Value(), &in); err != nil || in.Manifest == "" {
|
|
k.logger.Printf("%s.%s cannot be read, so its directory is read from the forge: %v", inputsBucket, tree, err)
|
|
return bridge.Inputs{}, false
|
|
}
|
|
k.mu.Lock()
|
|
k.mem[tree] = in
|
|
k.mu.Unlock()
|
|
return in, true
|
|
}
|
|
|
|
func (k *kept) Put(tree string, in bridge.Inputs) {
|
|
k.mu.Lock()
|
|
k.mem[tree] = in
|
|
k.mu.Unlock()
|
|
if k.kv == nil || !treeHash.MatchString(tree) {
|
|
return
|
|
}
|
|
data, err := json.Marshal(in)
|
|
if err != nil {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), keptTimeout)
|
|
defer cancel()
|
|
if _, err := k.kv.Put(ctx, tree, data); err != nil {
|
|
k.logger.Printf("%s.%s is not kept, so the next merge reads its directory from the forge again: %v", inputsBucket, tree, err)
|
|
}
|
|
}
|