Function TestKeyValueKeysDeduplicate

jetstream/test/kv_test.go:2082–2138

(t *testing.T)

}

func TestKeyValueKeysDeduplicate(t *testing.T) {
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc, js := jsClient(t, s)
	defer nc.Close()

	ctx := context.Background()
	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST_KV", History: 5})
	if err != nil {
		t.Fatalf("Error creating KV: %v", err)
	}

	for i := range 10 {
		key := fmt.Sprintf("key_%d", i)
		if _, err := kv.PutString(ctx, key, "initial"); err != nil {
			t.Fatalf("Error putting key %s: %v", key, err)
		}
	}

	done := make(chan bool)
	go func() {
		for {
			select {
			case <-done:
				return
			default:
				for i := range 5 {
					key := fmt.Sprintf("key_%d", i)
					if _, err := kv.PutString(ctx, key, "updated"); err != nil {
						if errors.Is(err, nats.ErrConnectionClosed) {
							return
						}
						t.Logf("Error updating key %s: %v", key, err)
					}
				}
			}
		}
	}()

	for range 20 {
		keys, err := kv.Keys(ctx)
		if err != nil {
			t.Fatalf("Error getting keys: %v", err)
		}

		seen := make(map[string]struct{})
		for _, key := range keys {
			if _, exists := seen[key]; exists {
				t.Fatalf("Duplicate key found: %s", key)
			}
			seen[key] = struct{}{}
		}
	}

	close(done)
}