diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index 1df0c665b29..ac8e9415101 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -387,21 +387,30 @@ func (b *backend) PrepareList(ctx context.Context, req *resource.ListRequest) (* } func (b *backend) WatchWriteEvents(ctx context.Context) (<-chan *resource.WrittenEvent, error) { + if err := b.Init(); err != nil { + return nil, err + } + // Fetch the lastest RV + rows, err := b.sqlDB.QueryContext(ctx, `SELECT COALESCE(max("resource_version"), 0) FROM "resource";`) + if err != nil { + return nil, fmt.Errorf("fetch latest rv: %w", err) + } + since := int64(0) + if rows.Next() { + if err := rows.Scan(&since); err != nil { + return nil, fmt.Errorf("scan since resource version: %w", err) + } + } + // Start the poller stream := make(chan *resource.WrittenEvent) - go b.poller(ctx, stream) + go b.poller(ctx, since, stream) return stream, nil } -func (b *backend) poller(ctx context.Context, stream chan<- *resource.WrittenEvent) { +func (b *backend) poller(ctx context.Context, since int64, stream chan<- *resource.WrittenEvent) { var err error - // FIXME: we need a way to state startup of server from a (Group, Resource) - // standpoint, and consider that new (Group, Resource) may be added to - // `kind_version`, so we should probably also poll for changes in there - since := int64(0) - interval := 100 * time.Millisecond // TODO make this configurable - t := time.NewTicker(interval) defer close(stream) defer t.Stop() @@ -433,7 +442,6 @@ func (b *backend) poll(ctx context.Context, since int64, stream chan<- *resource if err != nil { return 0, fmt.Errorf("execute SQL template to poll for resource history: %w", err) } - rows, err := b.sqlDB.QueryContext(ctx, query, pollReq.GetArgs()...) if err != nil { return 0, fmt.Errorf("poll for resource history: %w", err) diff --git a/pkg/storage/unified/sql/backend_test.go b/pkg/storage/unified/sql/backend_test.go index ebfa852d572..0cfd95df550 100644 --- a/pkg/storage/unified/sql/backend_test.go +++ b/pkg/storage/unified/sql/backend_test.go @@ -143,3 +143,48 @@ func TestBackendHappyPath(t *testing.T) { assert.Equal(t, resource.WatchEvent_DELETED, event.Type) }) } + +func TestBackendWatchWriteEventsFromHead(t *testing.T) { + ctx := context.Background() + dbstore := db.InitTestDB(t) + + rdb, err := dbimpl.ProvideResourceDB(dbstore, setting.NewCfg(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorage), nil) + assert.NoError(t, err) + store, err := NewBackendStore(backendOptions{ + DB: rdb, + }) + + assert.NoError(t, err) + assert.NotNil(t, store) + + // Create a few resources before initing the watch + _, err = store.WriteEvent(ctx, resource.WriteEvent{ + Type: resource.WatchEvent_ADDED, + Value: []byte("initial value 0"), + Key: &resource.ResourceKey{ + Namespace: "namespace", + Group: "group", + Resource: "resource", + Name: "item 0", + }, + }) + assert.NoError(t, err) + + // Start the watch + stream, err := store.WatchWriteEvents(ctx) + assert.NoError(t, err) + + // Create one more event + _, err = store.WriteEvent(ctx, resource.WriteEvent{ + Type: resource.WatchEvent_ADDED, + Value: []byte("initial value 2"), + Key: &resource.ResourceKey{ + Namespace: "namespace", + Group: "group", + Resource: "resource", + Name: "item2", + }, + }) + assert.NoError(t, err) + assert.Equal(t, "item2", (<-stream).Key.Name) +}