From bdf9298496c721916aaec6064a17fad2f5d9da69 Mon Sep 17 00:00:00 2001 From: Milas Bowman Date: Mon, 10 Mar 2025 22:54:48 -0400 Subject: [PATCH 1/2] chore(poll): return NotModifiedTimeoutErr on 304 --- client.go | 3 +++ poll_batch.go | 7 ++++++- poll_batch_test.go | 35 +++++++++++++++++++++++++++++++++++ poll_single_test.go | 21 +++++++++++++++++++++ 4 files changed, 65 insertions(+), 1 deletion(-) diff --git a/client.go b/client.go index 4bc6a1c..36a074b 100644 --- a/client.go +++ b/client.go @@ -22,6 +22,7 @@ const CausalityTokenHeader = "X-Garage-Causality-Token" var TombstoneItemErr = errors.New("item is a tombstone") var NoSuchItemErr = errors.New("item does not exist") var ConcurrentItemsErr = errors.New("item has multiple concurrent values") +var NotModifiedTimeoutErr = errors.New("not modified within timeout") var awsSigner = v4.NewSigner() @@ -300,6 +301,8 @@ func (c *Client) readItemSingle(ctx context.Context, b Bucket, pk string, sk str return nil, "", NoSuchItemErr case http.StatusConflict: return nil, ct, ConcurrentItemsErr + case http.StatusNotModified: + return nil, "", NotModifiedTimeoutErr default: return nil, "", fmt.Errorf("http status code %d", resp.StatusCode) } diff --git a/poll_batch.go b/poll_batch.go index c3ed20f..83df65b 100644 --- a/poll_batch.go +++ b/poll_batch.go @@ -67,7 +67,12 @@ func (c *Client) PollRange(ctx context.Context, b Bucket, pk string, q PollRange return nil, err } - if resp.StatusCode != http.StatusOK { + switch resp.StatusCode { + case http.StatusOK: + break + case http.StatusNotModified: + return nil, NotModifiedTimeoutErr + default: return nil, fmt.Errorf("http status code %d: %s", resp.StatusCode, body) } diff --git a/poll_batch_test.go b/poll_batch_test.go index 546bede..384aafa 100644 --- a/poll_batch_test.go +++ b/poll_batch_test.go @@ -80,3 +80,38 @@ func TestClient_PollRange(t *testing.T) { require.Equal(t, "hello3", string(result.Items[0].Values[0])) } } + +func TestClient_PollRange_Timeout(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode: 1 sec minimum to trigger timeout") + return + } + t.Parallel() + + f, ctx := newFixture(t) + + pk := randomPk() + sk := randomSk() + + for i := range 5 { + err := f.cli.InsertItem(ctx, f.bucket, pk, sk+"-"+strconv.Itoa(i), "", []byte("hello1")) + require.NoError(t, err) + } + + // first read should complete immediately + q := k2v.PollRangeQuery{ + Start: sk, + } + result, err := f.cli.PollRange(ctx, f.bucket, pk, q, 5*time.Second) + require.NoError(t, err) + require.NotEmpty(t, result.SeenMarker) + require.Len(t, result.Items, 5) + for i := range result.Items { + require.Len(t, result.Items[i].Values, 1) + require.Equal(t, "hello1", string(result.Items[i].Values[0])) + } + + q.SeenMarker = result.SeenMarker + result, err = f.cli.PollRange(ctx, f.bucket, pk, q, 1*time.Second) + require.ErrorIs(t, err, k2v.NotModifiedTimeoutErr) +} diff --git a/poll_single_test.go b/poll_single_test.go index 21f3276..dcca3de 100644 --- a/poll_single_test.go +++ b/poll_single_test.go @@ -46,3 +46,24 @@ func TestClient_PollItem(t *testing.T) { require.NotEmpty(t, ct) require.NoError(t, <-updateErrCh) } + +func TestClient_PollItem_Timeout(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode: 1 sec minimum to trigger timeout") + return + } + t.Parallel() + + f, ctx := newFixture(t) + + pk := randomPk() + sk := randomSk() + + err := f.cli.InsertItem(ctx, f.bucket, pk, sk, "", []byte("hello1")) + require.NoError(t, err) + + _, ct, err := f.cli.ReadItemSingle(ctx, f.bucket, pk, sk) + item, ct, err := f.cli.PollItem(ctx, f.bucket, pk, sk, ct, 1*time.Second) + require.ErrorIs(t, err, k2v.NotModifiedTimeoutErr) + require.Empty(t, item) +} From f47bdf7fd2d47ae6036579c580e5438ad32a95cc Mon Sep 17 00:00:00 2001 From: Milas Bowman Date: Fri, 28 Mar 2025 21:50:12 -0400 Subject: [PATCH 2/2] fix: awsv4 signatures + http client --- client.go | 25 ++++++++++++++++++------- client_test.go | 2 +- main_test.go | 8 +++++++- pager_test.go | 5 ++--- poll_batch_test.go | 1 + poll_single_test.go | 5 ++++- read_batch.go | 6 +++--- 7 files changed, 36 insertions(+), 16 deletions(-) diff --git a/client.go b/client.go index 36a074b..74f24ad 100644 --- a/client.go +++ b/client.go @@ -19,6 +19,9 @@ import ( const CausalityTokenHeader = "X-Garage-Causality-Token" +const payloadHashEmpty = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" +const payloadHashUnsigned = "UNSIGNED-PAYLOAD" + var TombstoneItemErr = errors.New("item is a tombstone") var NoSuchItemErr = errors.New("item does not exist") var ConcurrentItemsErr = errors.New("item has multiple concurrent values") @@ -95,7 +98,7 @@ func (c *Client) executeRequest(req *http.Request) (*http.Response, error) { return nil, err } - resp, err := http.DefaultClient.Do(req) + resp, err := c.httpClient.Do(req) if err != nil { return nil, err } @@ -104,14 +107,25 @@ func (c *Client) executeRequest(req *http.Request) (*http.Response, error) { } func (c *Client) signRequest(req *http.Request) error { + if c.key.ID == "" || c.key.Secret == "" { + return errors.New("no credentials provided") + } + creds := aws.Credentials{ AccessKeyID: c.key.ID, SecretAccessKey: c.key.Secret, } - const noBody = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" - req.Header.Set("X-Amz-Content-Sha256", noBody) - err := awsSigner.SignHTTP(req.Context(), creds, req, noBody, "k2v", "garage", time.Now()) + var payloadHash string + if req.Body == nil || req.Body == http.NoBody { + + payloadHash = payloadHashEmpty + } else { + payloadHash = payloadHashUnsigned + } + req.Header.Set("X-Amz-Content-Sha256", payloadHash) + + err := awsSigner.SignHTTP(req.Context(), creds, req, payloadHash, "k2v", "garage", time.Now()) if err != nil { return err } @@ -248,9 +262,6 @@ func (c *Client) ReadItemMulti(ctx context.Context, b Bucket, pk string, sk stri return []Item{body}, ct, nil case "application/json": var items []Item - if err != nil { - return nil, "", err - } if err := json.Unmarshal(body, &items); err != nil { return nil, "", err } diff --git a/client_test.go b/client_test.go index 2aa0605..d7feaf1 100644 --- a/client_test.go +++ b/client_test.go @@ -27,7 +27,7 @@ func newFixture(t testing.TB) (*fixture, context.Context) { t: t, ctx: ctx, cli: cli, - bucket: k2v.Bucket("k2v-test"), + bucket: TestBucket, } return f, ctx diff --git a/main_test.go b/main_test.go index c5a3d85..65554a0 100644 --- a/main_test.go +++ b/main_test.go @@ -1,10 +1,16 @@ -package k2v +package k2v_test import ( + k2v "code.notaphish.fyi/milas/garage-k2v-go" "go.uber.org/goleak" + "os" "testing" ) +const BucketEnvVar = "K2V_TEST_BUCKET" + +var TestBucket = k2v.Bucket(os.Getenv(BucketEnvVar)) + func TestMain(m *testing.M) { goleak.VerifyTestMain(m) } diff --git a/pager_test.go b/pager_test.go index 5d5a661..6ed8cd6 100644 --- a/pager_test.go +++ b/pager_test.go @@ -32,15 +32,14 @@ func ExampleScrollIndex() { ctx := context.Background() client := k2v.NewClient(k2v.EndpointFromEnv(), k2v.KeyFromEnv()) defer client.Close() - const bucket = "k2v-test" pkPrefix := randomPk() for i := range 5 { - _ = client.InsertItem(ctx, bucket, pkPrefix+"-"+strconv.Itoa(i), randomSk(), "", []byte("hello")) + _ = client.InsertItem(ctx, TestBucket, pkPrefix+"-"+strconv.Itoa(i), randomSk(), "", []byte("hello")) } var responses []*k2v.ReadIndexResponse - _ = k2v.ScrollIndex(ctx, client, bucket, k2v.ReadIndexQuery{Prefix: pkPrefix, Limit: 25}, func(resp *k2v.ReadIndexResponse) error { + _ = k2v.ScrollIndex(ctx, client, TestBucket, k2v.ReadIndexQuery{Prefix: pkPrefix, Limit: 25}, func(resp *k2v.ReadIndexResponse) error { responses = append(responses, resp) return nil }) diff --git a/poll_batch_test.go b/poll_batch_test.go index 384aafa..c0c91cd 100644 --- a/poll_batch_test.go +++ b/poll_batch_test.go @@ -114,4 +114,5 @@ func TestClient_PollRange_Timeout(t *testing.T) { q.SeenMarker = result.SeenMarker result, err = f.cli.PollRange(ctx, f.bucket, pk, q, 1*time.Second) require.ErrorIs(t, err, k2v.NotModifiedTimeoutErr) + require.Nil(t, result) } diff --git a/poll_single_test.go b/poll_single_test.go index dcca3de..d65521e 100644 --- a/poll_single_test.go +++ b/poll_single_test.go @@ -20,6 +20,7 @@ func TestClient_PollItem(t *testing.T) { require.NoError(t, err) _, ct, err := f.cli.ReadItemSingle(ctx, f.bucket, pk, sk) + require.NoError(t, err) updateErrCh := make(chan error, 1) pollReadyCh := make(chan struct{}) @@ -63,7 +64,9 @@ func TestClient_PollItem_Timeout(t *testing.T) { require.NoError(t, err) _, ct, err := f.cli.ReadItemSingle(ctx, f.bucket, pk, sk) - item, ct, err := f.cli.PollItem(ctx, f.bucket, pk, sk, ct, 1*time.Second) + require.NoError(t, err) + + item, _, err := f.cli.PollItem(ctx, f.bucket, pk, sk, ct, 1*time.Second) require.ErrorIs(t, err, k2v.NotModifiedTimeoutErr) require.Empty(t, item) } diff --git a/read_batch.go b/read_batch.go index 2ff045c..e4382a5 100644 --- a/read_batch.go +++ b/read_batch.go @@ -54,9 +54,9 @@ type BatchSearchResult struct { } type SearchResultItem struct { - SortKey string `json:"sk"` - CausalityToken string `json:"ct"` - Values []Item `json:"v"` + SortKey string `json:"sk"` + CausalityToken CausalityToken `json:"ct"` + Values []Item `json:"v"` } func (c *Client) ReadBatch(ctx context.Context, b Bucket, q []BatchSearch) ([]*BatchSearchResult, error) {