Skip to content

[LFXV2-2190, LFXV2-2193] fix: lazy-bind weekly-brief KV buckets and fix query-service version param #167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andrest50 merged 4 commits into from
Aug 3, 2026
Merged

[LFXV2-2190, LFXV2-2193] fix: lazy-bind weekly-brief KV buckets and fix query-service version param #167

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8d5f8ba
fix(LFXV2-2190): lazy-bind weekly-brief KV buckets on first access
andrest50 Aug 3, 2026
c914bb8
fix(LFXV2-2193): add version=1 query param to all m2m source clients
andrest50 Aug 3, 2026
b3344a4
fix(review): isolate lazy KV binds and move bind call outside lock
andrest50 Aug 3, 2026
7bef1f8
fix(review): use correct HTTP param name v=1 for query-service version
andrest50 Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 34 additions & 21 deletions internal/infrastructure/nats/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"context"
"fmt"
"log/slog"
"sync"
"time"

"github.com/linuxfoundation/lfx-v2-committee-service/internal/domain/port"
Expand All @@ -25,6 +26,7 @@ import (
type NATSClient struct {
conn *nats.Conn
config Config
kvMu sync.RWMutex
kvStore map[string]jetstream.KeyValue
objStore map[string]jetstream.ObjectStore
timeout time.Duration
Expand Down Expand Up @@ -83,6 +85,38 @@ func (c *NATSClient) KeyValueStore(ctx context.Context, bucketName string) error
return nil
}

// GetOrBindKVStore returns the KV handle for bucketName. If the bucket was
// absent at startup (e.g. a pod that booted before the bucket was provisioned),
// it attempts to bind on first access so the pod self-heals without a restart.
func (c *NATSClient) GetOrBindKVStore(ctx context.Context, bucketName string) (jetstream.KeyValue, error) {
c.kvMu.RLock()
kv, ok := c.kvStore[bucketName]
c.kvMu.RUnlock()
if ok {
return kv, nil
}

c.kvMu.Lock()
defer c.kvMu.Unlock()
if kv, ok = c.kvStore[bucketName]; ok {
return kv, nil
}
js, err := jetstream.New(c.conn)
if err != nil {
return nil, errors.NewServiceUnavailable("failed to create JetStream client", err)
}
kv, err = js.KeyValue(ctx, bucketName)
if err != nil {
return nil, errors.NewServiceUnavailable(fmt.Sprintf("%s bucket not initialized", bucketName), err)
}
if c.kvStore == nil {
c.kvStore = make(map[string]jetstream.KeyValue)
}
c.kvStore[bucketName] = kv
Comment thread
andrest50 marked this conversation as resolved.
Outdated
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
slog.InfoContext(ctx, "weekly-brief KV bucket bound on first access", "bucket", bucketName)
return kv, nil
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

// ObjectStore creates a JetStream client and gets the object store by name.
func (c *NATSClient) ObjectStore(ctx context.Context, storeName string) error {
js, err := jetstream.New(c.conn)
Expand Down Expand Up @@ -272,27 +306,6 @@ func NewClient(ctx context.Context, config Config) (*NATSClient, error) {
)
}

// Weekly-brief buckets are initialized best-effort. If they aren't yet
// provisioned (e.g. a rolling deploy where the chart hasn't created them, or
// a local NATS without them) the service still starts; only the weekly-brief
// endpoints return ServiceUnavailable until the buckets exist.
for _, bucketName := range []string{
constants.KVBucketNameGroupWeeklyBriefs,
constants.KVBucketNameGroupWeeklyBriefUIDIndex,
constants.KVBucketNameGroupWeeklyBriefThrottle,
} {
if err := client.KeyValueStore(ctx, bucketName); err != nil {
slog.WarnContext(ctx, "weekly-brief KV bucket not initialized; weekly-brief endpoints will be unavailable until it is provisioned",
"error", err,
"bucket", bucketName,
)
continue
}
slog.InfoContext(ctx, "NATS key-value store initialized",
"bucket", bucketName,
)
}

for _, storeName := range []string{
constants.ObjectStoreNameCommitteeDocuments,
} {
Expand Down
38 changes: 19 additions & 19 deletions internal/infrastructure/nats/group_weekly_brief_storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,9 +47,9 @@ func (s *storage) GetGroupWeeklyBriefForWindow(ctx context.Context, committeeUID
// WindowStart; the rest of the fields are unused.
indexKey := buildBriefIndexKey(committeeUID, model.WindowDateKey(windowStart.WindowStart))

idxBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefUIDIndex]
if !ok {
return nil, nil, errs.NewServiceUnavailable("group-weekly-brief-uid-index bucket not initialized")
idxBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefUIDIndex)
if err != nil {
return nil, nil, err
}

entry, err := idxBucket.Get(ctx, indexKey)
Expand All @@ -64,9 +64,9 @@ func (s *storage) GetGroupWeeklyBriefForWindow(ctx context.Context, committeeUID
return nil, nil, nil
}

briefBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefs]
if !ok {
return nil, nil, errs.NewServiceUnavailable("group-weekly-briefs bucket not initialized")
briefBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefs)
if err != nil {
return nil, nil, err
}
briefEntry, errGet := briefBucket.Get(ctx, sanitizeKVKey(briefUID))
if errGet != nil {
Expand Down Expand Up @@ -107,7 +107,7 @@ func (s *storage) GetGroupWeeklyBriefForWindow(ctx context.Context, committeeUID
// Best-effort throttle lookup. Misses and errors don't fail the read —
// throttle is advisory metadata.
var throttleBytes []byte
if thBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefThrottle]; ok {
if thBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefThrottle); err == nil {
thEntry, thErr := thBucket.Get(ctx, indexKey)
switch {
case thErr == nil:
Expand Down Expand Up @@ -153,13 +153,13 @@ func (s *storage) PutGroupWeeklyBrief(ctx context.Context, brief *model.GroupWee
}
brief.UpdatedAt = now

briefBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefs]
if !ok {
return nil, errs.NewServiceUnavailable("group-weekly-briefs bucket not initialized")
briefBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefs)
if err != nil {
return nil, err
}
idxBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefUIDIndex]
if !ok {
return nil, errs.NewServiceUnavailable("group-weekly-brief-uid-index bucket not initialized")
idxBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefUIDIndex)
if err != nil {
return nil, err
}

payload, err := json.Marshal(brief)
Expand Down Expand Up @@ -211,9 +211,9 @@ func (s *storage) PutGroupWeeklyBrief(ctx context.Context, brief *model.GroupWee
// GetGroupWeeklyBriefThrottle returns the throttle entry for the given
// (committee, window-start) pair. A miss returns (nil, nil).
func (s *storage) GetGroupWeeklyBriefThrottle(ctx context.Context, committeeUID string, windowStart time.Time) (*model.GroupWeeklyBriefThrottle, error) {
thBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefThrottle]
if !ok {
return nil, errs.NewServiceUnavailable("group-weekly-brief-throttle bucket not initialized")
thBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefThrottle)
if err != nil {
return nil, err
}
key := buildBriefIndexKey(committeeUID, model.WindowDateKey(windowStart))
entry, err := thBucket.Get(ctx, key)
Expand Down Expand Up @@ -245,9 +245,9 @@ func (s *storage) PutGroupWeeklyBriefThrottle(ctx context.Context, throttle *mod
if throttle.WindowStart.IsZero() {
return nil, errs.NewValidation("window_start is required")
}
thBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefThrottle]
if !ok {
return nil, errs.NewServiceUnavailable("group-weekly-brief-throttle bucket not initialized")
thBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefThrottle)
if err != nil {
return nil, err
}

payload, err := json.Marshal(throttle)
Expand Down
Loading