3pp-mirror/pomerium
1
0
Fork 0
mirror of https://github.com/pomerium/pomerium.git synced 2025-04-28 18:06:34 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix record version in query response

Browse source
This commit is contained in:
Caleb Doxsey 2025-04-10 10:32:04 -06:00
parent dd62c32739
commit 66178c1fbe
4 changed files with 22 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
pkg/storage/querier.go
View file

@ -45,11 +45,6 @@ func WithQuerier(ctx context.Context, querier Querier) context.Context {
return context.WithValue(ctx, querierKey{}, querier)
}
// Query queries for records.
func (q *staticQuerier) Query(_ context.Context, req *databroker.QueryRequest, _ ...grpc.CallOption) (*databroker.QueryResponse, error) {
return QueryRecordCollections(q.records, req)
}
// MarshalQueryRequest marshales the query request.
func MarshalQueryRequest(req *databroker.QueryRequest) ([]byte, error) {
return (&proto.MarshalOptions{

20
pkg/storage/querier_caching_test.go
View file

@ -36,8 +36,9 @@ func TestCachingQuerier(t *testing.T) {
})
assert.NoError(t, err)
assert.Empty(t, cmp.Diff(&databrokerpb.QueryResponse{
Records: []*databrokerpb.Record{{Version: 1, Type: "t1", Id: "r1"}},
TotalCount: 1,
Records: []*databrokerpb.Record{{Version: 1, Type: "t1", Id: "r1"}},
TotalCount: 1,
RecordVersion: 1,
}, res, protocmp.Transform()))
res, err = storage.NewCachingQuerier(q2, cache).Query(ctx, &databrokerpb.QueryRequest{
@ -46,8 +47,9 @@ func TestCachingQuerier(t *testing.T) {
})
assert.NoError(t, err)
assert.Empty(t, cmp.Diff(&databrokerpb.QueryResponse{
Records: []*databrokerpb.Record{{Version: 1, Type: "t1", Id: "r1"}},
TotalCount: 1,
Records: []*databrokerpb.Record{{Version: 1, Type: "t1", Id: "r1"}},
TotalCount: 1,
RecordVersion: 1,
}, res, protocmp.Transform()), "should use the cached version")
res, err = storage.NewCachingQuerier(q2, cache).Query(ctx, &databrokerpb.QueryRequest{
@ -57,8 +59,9 @@ func TestCachingQuerier(t *testing.T) {
})
assert.NoError(t, err)
assert.Empty(t, cmp.Diff(&databrokerpb.QueryResponse{
Records: []*databrokerpb.Record{{Version: 1, Type: "t1", Id: "r1"}},
TotalCount: 1,
Records: []*databrokerpb.Record{{Version: 1, Type: "t1", Id: "r1"}},
TotalCount: 1,
RecordVersion: 1,
}, res, protocmp.Transform()), "should use the cached version")
res, err = storage.NewCachingQuerier(q2, cache).Query(ctx, &databrokerpb.QueryRequest{
@ -68,7 +71,8 @@ func TestCachingQuerier(t *testing.T) {
})
assert.NoError(t, err)
assert.Empty(t, cmp.Diff(&databrokerpb.QueryResponse{
Records: []*databrokerpb.Record{{Version: 2, Type: "t1", Id: "r1"}},
TotalCount: 1,
Records: []*databrokerpb.Record{{Version: 2, Type: "t1", Id: "r1"}},
TotalCount: 1,
RecordVersion: 2,
}, res, protocmp.Transform()), "should query the new version")
}

6
pkg/storage/querier_static.go
View file

@ -6,6 +6,7 @@ import (
"strconv"
"github.com/google/uuid"
grpc "google.golang.org/grpc"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
timestamppb "google.golang.org/protobuf/types/known/timestamppb"
@ -75,3 +76,8 @@ func NewStaticRecord(typeURL string, msg proto.Message) *databroker.Record {
}
func (q *staticQuerier) InvalidateCache(_ context.Context, _ *databroker.QueryRequest) {}
// Query queries for records.
func (q *staticQuerier) Query(_ context.Context, req *databroker.QueryRequest, _ ...grpc.CallOption) (*databroker.QueryResponse, error) {
return QueryRecordCollections(q.records, req)
}

4
pkg/storage/record_collection.go
View file

@ -328,6 +328,10 @@ func QueryRecordCollections(
res := new(databroker.QueryResponse)
for _, c := range cs {
if record, ok := c.Newest(); ok {
res.RecordVersion = max(res.RecordVersion, record.Version)
}
records, err := c.List(filter)
if err != nil {
return nil, err