internal/gerrit/sync.go - oscar - Git at Google

 // Copyright 2024 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 // Package gerrit mirrors Gerrit CL state in a [storage.DB].
 package gerrit

 import (
 	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"iter"
 	"log/slog"
 	"net/http"
 	"net/url"
 	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"

 	"golang.org/x/oscar/internal/secret"
 	"golang.org/x/oscar/internal/storage"
 	"golang.org/x/oscar/internal/storage/timed"
 	"rsc.io/ordered"
 )

 const (
 	syncProjectKind  = "gerrit.SyncProject"
 	changeKind       = "gerrit.Change"
 	commentKind      = "gerrit.Comment"
 	changeUpdateKind = "gerrit.ChangeUpdate"
 )

 // For a Gerrit project we store changes indexed by change number.
 // For each change we store the latest known state, as the JSON
 // encoding of what Gerrit calls a ChangeInfo entity.
 // We also store comments on the change, a JSON array of Gerrit
 // CommentInfo entities.
 //
 // We store a simple timed stream of updated change numbers.
 // We don't store historical status of changes, as all historical
 // information is stored in the ChangeInfo and CommentInto entities.
 //
 // The following key schemas are stored in the database:
 //
 //	["gerritSyncProject", Instance, Project] => JSON of projectSync structure
 //	["gerrit.Change", Instance, Project, ChangeNumber] => ChangeInfo JSON
 //	["gerrit.Comment", Instance, Project, ChangeNumber] => CommentInfo JSON
 //	["gerrit.ChangeUpdate", Instance, ChangeNumber, MetaID] => DBTime
 //	["gerrit.ChangeUpdateByTime", DBTime, Instance, ChangeNumber, MetaID] => []
 //
 // A watcher on "gerrit.ChangeUpdate" will see all Gerrit changes,
 // and can read the new data from the database.

 // changeOpts is the options we request for a change.
 var changeOpts = []string{
 	"ALL_REVISIONS",
 	"DETAILED_ACCOUNTS",
 	"LABELS",
 	"ALL_COMMITS",
 	"MESSAGES",
 	"SUBMITTABLE",
 	"PARENTS",
 }

 // o is short for ordered.Encode.
 func o(list ...any) []byte { return ordered.Encode(list...) }

 // A Client is a connection to a Gerrit instance, and to the database
 // that stores information gathered from the instance.
 type Client struct {
 	instance string
 	slog     *slog.Logger
 	db       storage.DB
 	secret   secret.DB
 	http     *http.Client

 	flushRequested atomic.Bool // flush database to disk when convenient

 	testing bool
 }

 // New returns a new client to access a Gerrit instance
 // described by a host name like "go-review.googlesource.com".
 // The client uses the given logger, databases, and HTTP client.
 //
 // The secret database will look for a secret whose name is the
 // Gerrit instance. The value will be user:pass.  This is not yet used.
 func New(instance string, lg *slog.Logger, db storage.DB, sdb secret.DB, hc *http.Client) *Client {
 	return &Client{
 		instance: instance,
 		slog:     lg,
 		db:       db,
 		secret:   sdb,
 		http:     hc,
 		testing:  testing.Testing(),
 	}
 }

 // RequestFlush asks a Gerrit sync to flush the database to disk
 // when convenient. This may be called concurrently with Sync.
 func (c *Client) RequestFlush() {
 	c.flushRequested.Store(true)
 }

 // projectSync records the sync state of a Gerrit project within
 // an instance, such as "go" or "website".
 // This is stored in the database.
 type projectSync struct {
 	Instance string // instance host name, "go-review.googlesource.com"
 	Name     string // project name, such as "go" or "oscar".
 	SyncDate string // date/time of last sync
 }

 // store stores inst into db.
 func (proj *projectSync) store(db storage.DB) {
 	db.Set(o(syncProjectKind, proj.Instance, proj.Name), storage.JSON(proj))
 }

 // Add adds a Gerrit project such as "go" or "oscar" to the database.
 // It only adds the project sync metadata.
 // The initial data fetch does not happen until [Sync] or [SyncProject]
 // is called.
 // Add returns an error if the project has already been added.
 func (c *Client) Add(project string) error {
 	key := o(syncProjectKind, c.instance, project)
 	if _, ok := c.db.Get(key); ok {
 		return fmt.Errorf("gerrit.Add: already added: %q", project)
 	}
 	proj := &projectSync{
 		Instance: c.instance,
 		Name:     project,
 	}
 	c.db.Set(key, storage.JSON(proj))
 	return nil
 }

 // Sync syncs the data for all projects in this client's instance.
 func (c *Client) Sync(ctx context.Context) error {
 	var errs []error
 	for key := range c.db.Scan(o(syncProjectKind, c.instance), o(syncProjectKind, c.instance, ordered.Inf)) {
 		var project string
 		if err := ordered.Decode(key, nil, nil, &project); err != nil {
 			c.db.Panic("gerrit client sync decode", "key", storage.Fmt(key), "err", err)
 		}
 		if err := c.SyncProject(ctx, project); err != nil {
 			errs = append(errs, err)
 		}
 	}
 	return errors.Join(errs...)
 }

 // SyncProject syncs a single project.
 func (c *Client) SyncProject(ctx context.Context, project string) (err error) {
 	c.slog.Debug("gerrit.SyncProject", "project", project)
 	defer func() {
 		if err != nil {
 			err = fmt.Errorf("SyncProject(%q): %w", project, err)
 		}
 	}()

 	key := o(syncProjectKind, c.instance, project)
 	skey := string(key)

 	// Lock the project, so that no else is sync'ing concurrently.
 	c.db.Lock(skey)
 	defer c.db.Unlock(skey)

 	// Load sync state.
 	var proj projectSync
 	if val, ok := c.db.Get(key); !ok {
 		return errors.New("missing project")
 	} else if err := json.Unmarshal(val, &proj); err != nil {
 		return err
 	}

 	return c.syncChanges(ctx, &proj)
 }

 // syncChanges finds all changes that have been updated since the last
 // sync time. It stores the data for those changes in the database.
 // It also adds in the metadata changes.
 func (c *Client) syncChanges(ctx context.Context, proj *projectSync) (err error) {
 	b := c.db.Batch()
 	defer b.Apply()

 	// When we need to fetch multiple lists of changes,
 	// concurrent modifications can cause us to see the
 	// same change more than once. Keep track of the changes
 	// we've already seen.
 	seen := make(map[int]json.RawMessage)

 	// We see the changes from newest to oldest.
 	// Update the sync time if we get through
 	// all of them with no error.
 	lastUpdate := ""
 	defer func() {
 		if err == nil && lastUpdate != "" {
 			proj.SyncDate = lastUpdate
 			proj.store(c.db)
 		}
 	}()

 	for change, err := range c.changes(ctx, proj.Name, proj.SyncDate) {
 		if err != nil {
 			return err
 		}
 		if err := ctx.Err(); err != nil {
 			return err
 		}

 		if c.flushRequested.Load() {
 			// Flush database.
 			b.Apply()
 			c.db.Flush()
 			c.flushRequested.Store(false)
 		}

 		if lastUpdate == "" {
 			// Save the update time of the most recent
 			// change to record in the project sync.
 			var updated struct {
 				Updated string `json:"updated"`
 			}
 			if err := json.Unmarshal(change, &updated); err != nil {
 				return err
 			}
 			lastUpdate = updated.Updated
 		}

 		// Change is a Gerrit ChangeInfo in JSON form.
 		// Pull out the change number.
 		var num struct {
 			Number int    `json:"_number"`
 			MetaID string `json:"meta_rev_id"`
 		}
 		if err := json.Unmarshal(change, &num); err != nil {
 			return err
 		}
 		changeNum := num.Number
 		metaID := num.MetaID
 		if changeNum == 0 {
 			return fmt.Errorf("missing _number field in %q", change)
 		}
 		if metaID == "" {
 			return fmt.Errorf("missing meta_rev_id field in change %d: %q", changeNum, change)
 		}

 		if oldChange, ok := seen[changeNum]; ok {
 			same, err := sameChangeInfo(change, metaID, oldChange)
 			if err != nil {
 				return err
 			}
 			if same {
 				// Nothing has changed.
 				continue
 			}
 		}

 		seen[changeNum] = change

 		key := o(changeKind, c.instance, proj.Name, changeNum)
 		if oldChange, ok := c.db.Get(key); ok {
 			same, err := sameChangeInfo(change, metaID, oldChange)
 			if err != nil {
 				return err
 			}
 			if same {
 				// Nothing has changed.
 				continue
 			}
 		}

 		b.Set(key, change)
 		if err := c.syncComments(ctx, b, proj.Name, changeNum); err != nil {
 			return err
 		}

 		// Record that the change was updated.
 		timed.Set(c.db, b, changeUpdateKind, o(c.instance, changeNum, metaID), nil)

 		b.MaybeApply()
 	}

 	return nil
 }

 // syncComments updates the comments of a change in the database.
 func (c *Client) syncComments(ctx context.Context, b storage.Batch, project string, changeNum int) error {
 	url := "https://" + c.instance + "/changes/" + strconv.Itoa(changeNum) + "/comments"
 	var obj json.RawMessage
 	if err := c.get(ctx, url, &obj); err != nil {
 		return err
 	}

 	key := o(commentKind, c.instance, project, changeNum)
 	b.Set(key, obj)
 	return nil
 }

 // testBefore exists for testing purposes, to only pull changes before
 // some time. For ordinary use this should be the empty string.
 var testBefore string

 // changes returns an iterator over changes in the Gerrit repo.
 // If date is not the empty string, changes only returns changes
 // that were updated after date.
 func (c *Client) changes(ctx context.Context, project, date string) iter.Seq2[json.RawMessage, error] {
 	return func(yield func(json.RawMessage, error) bool) {
 		baseURL := "https://" + c.instance + "/changes"

 		// Gerrit returns up to 500 changes at a time.
 		// Gerrit provides a way to skip leading changes,
 		// by passing an S parameter.
 		// Unfortunately, S only works up to a value of 10,000.
 		// Beyond that Gerrit returns no changes.
 		// So to get all the changes we have to ask for changes
 		// before some time.
 		before := testBefore

 		for {
 			values := url.Values{
 				"o": changeOpts,
 			}
 			query := "p:" + project
 			if date != "" {
 				query += " after:" + date
 			}
 			if before != "" {
 				query += " before:" + before
 			}
 			values.Set("q", query)
 			addr := baseURL + "?" + values.Encode()

 			var body []json.RawMessage
 			if err := c.get(ctx, addr, &body); err != nil {
 				yield(nil, err)
 				return
 			}

 			if len(body) == 0 {
 				return
 			}

 			// If the last element in body has a
 			// _more_changes field, then skip it this time,
 			// and pick it up next time with a before query.
 			var more struct {
 				MoreChanges bool   `json:"_more_changes"`
 				Updated     string `json:"updated"`
 			}
 			last := body[len(body)-1]
 			if err := json.Unmarshal(last, &more); err != nil {
 				yield(nil, err)
 				return
 			}
 			if more.MoreChanges {
 				if len(body) == 1 {
 					yield(nil, errors.New("single change has _more_changes set"))
 				}
 				body = body[:len(body)-1]
 			}

 			for _, change := range body {
 				if !yield(change, nil) {
 					return
 				}
 			}

 			if !more.MoreChanges {
 				return
 			}

 			// Fortunately the before query is inclusive.
 			before = more.Updated
 		}
 	}
 }

 // sameChangeInfo reports whether two ChangeInfo structures,
 // in JSON form, are the same. aMetaID is the meta ID of a.
 func sameChangeInfo(a []byte, aMetaID string, b []byte) (bool, error) {
 	if bytes.Equal(a, b) {
 		return true, nil
 	}

 	// Unfortunately Gerrit does not return identical ChangeInfo
 	// information with consistent field ordering.
 	// In particular, we've seen that the order of "reviewers"
 	// can change. So we check the meta ID.
 	var extractMetaID struct {
 		MetaID string `json:"meta_rev_id"`
 	}

 	if err := json.Unmarshal(b, &extractMetaID); err != nil {
 		return false, err
 	}
 	bMetaID := extractMetaID.MetaID
 	if bMetaID == "" {
 		return false, errors.New("missing meta ID")
 	}

 	return aMetaID == bMetaID, nil
 }

 // get fetches addr and decodes the body as JSON into obj.
 func (c *Client) get(ctx context.Context, addr string, obj any) error {
 	c.slog.Info("gerrit GET", "addr", addr)

 	tries := 0
 	backoff := 1 * time.Second
 	for {
 		req, err := http.NewRequestWithContext(ctx, "GET", addr, nil)
 		if err != nil {
 			return err
 		}
 		resp, err := c.http.Do(req)
 		if err != nil {
 			return err
 		}

 		if resp.StatusCode != http.StatusOK {
 			data, err := io.ReadAll(resp.Body)
 			resp.Body.Close()
 			if err != nil {
 				return fmt.Errorf("reading body: %v", err)
 			}

 			if resp.StatusCode == http.StatusTooManyRequests {
 				tries++
 				if tries > 20 {
 					return errors.New("too many requests")
 				}
 				c.slog.Info("gerrit too many requests",
 					"try", tries,
 					"sleep", backoff,
 					"body", string(data))
 				time.Sleep(backoff)
 				backoff = min(backoff*2, 1*time.Minute)

 				continue
 			}

 			return fmt.Errorf("%s\n%s", resp.Status, data)
 		}

 		// Skip the XSRF header at the start of the response.
 		buf := bufio.NewReader(resp.Body)
 		defer resp.Body.Close()
 		if _, err := buf.ReadSlice('\n'); err != nil {
 			return err
 		}

 		return json.NewDecoder(buf).Decode(obj)
 	}
 }
	// Copyright 2024 The Go Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	// Package gerrit mirrors Gerrit CL state in a [storage.DB].
	package gerrit

	import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"iter"
	"log/slog"
	"net/http"
	"net/url"
	"strconv"
	"sync/atomic"
	"testing"
	"time"

	"golang.org/x/oscar/internal/secret"
	"golang.org/x/oscar/internal/storage"
	"golang.org/x/oscar/internal/storage/timed"
	"rsc.io/ordered"
	)

	const (
	syncProjectKind = "gerrit.SyncProject"
	changeKind = "gerrit.Change"
	commentKind = "gerrit.Comment"
	changeUpdateKind = "gerrit.ChangeUpdate"
	)

	// For a Gerrit project we store changes indexed by change number.
	// For each change we store the latest known state, as the JSON
	// encoding of what Gerrit calls a ChangeInfo entity.
	// We also store comments on the change, a JSON array of Gerrit
	// CommentInfo entities.
	//
	// We store a simple timed stream of updated change numbers.
	// We don't store historical status of changes, as all historical
	// information is stored in the ChangeInfo and CommentInto entities.
	//
	// The following key schemas are stored in the database:
	//
	// ["gerritSyncProject", Instance, Project] => JSON of projectSync structure
	// ["gerrit.Change", Instance, Project, ChangeNumber] => ChangeInfo JSON
	// ["gerrit.Comment", Instance, Project, ChangeNumber] => CommentInfo JSON
	// ["gerrit.ChangeUpdate", Instance, ChangeNumber, MetaID] => DBTime
	// ["gerrit.ChangeUpdateByTime", DBTime, Instance, ChangeNumber, MetaID] => []
	//
	// A watcher on "gerrit.ChangeUpdate" will see all Gerrit changes,
	// and can read the new data from the database.

	// changeOpts is the options we request for a change.
	var changeOpts = []string{
	"ALL_REVISIONS",
	"DETAILED_ACCOUNTS",
	"LABELS",
	"ALL_COMMITS",
	"MESSAGES",
	"SUBMITTABLE",
	"PARENTS",
	}

	// o is short for ordered.Encode.
	func o(list ...any) []byte { return ordered.Encode(list...) }

	// A Client is a connection to a Gerrit instance, and to the database
	// that stores information gathered from the instance.
	type Client struct {
	instance string
	slog *slog.Logger
	db storage.DB
	secret secret.DB
	http *http.Client

	flushRequested atomic.Bool // flush database to disk when convenient

	testing bool
	}

	// New returns a new client to access a Gerrit instance
	// described by a host name like "go-review.googlesource.com".
	// The client uses the given logger, databases, and HTTP client.
	//
	// The secret database will look for a secret whose name is the
	// Gerrit instance. The value will be user:pass. This is not yet used.
	func New(instance string, lg slog.Logger, db storage.DB, sdb secret.DB, hc http.Client) *Client {
	return &Client{
	instance: instance,
	slog: lg,
	db: db,
	secret: sdb,
	http: hc,
	testing: testing.Testing(),
	}
	}

	// RequestFlush asks a Gerrit sync to flush the database to disk
	// when convenient. This may be called concurrently with Sync.
	func (c *Client) RequestFlush() {
	c.flushRequested.Store(true)
	}

	// projectSync records the sync state of a Gerrit project within
	// an instance, such as "go" or "website".
	// This is stored in the database.
	type projectSync struct {
	Instance string // instance host name, "go-review.googlesource.com"
	Name string // project name, such as "go" or "oscar".
	SyncDate string // date/time of last sync
	}

	// store stores inst into db.
	func (proj *projectSync) store(db storage.DB) {
	db.Set(o(syncProjectKind, proj.Instance, proj.Name), storage.JSON(proj))
	}

	// Add adds a Gerrit project such as "go" or "oscar" to the database.
	// It only adds the project sync metadata.
	// The initial data fetch does not happen until [Sync] or [SyncProject]
	// is called.
	// Add returns an error if the project has already been added.
	func (c *Client) Add(project string) error {
	key := o(syncProjectKind, c.instance, project)
	if _, ok := c.db.Get(key); ok {
	return fmt.Errorf("gerrit.Add: already added: %q", project)
	}
	proj := &projectSync{
	Instance: c.instance,
	Name: project,
	}
	c.db.Set(key, storage.JSON(proj))
	return nil
	}

	// Sync syncs the data for all projects in this client's instance.
	func (c *Client) Sync(ctx context.Context) error {
	var errs []error
	for key := range c.db.Scan(o(syncProjectKind, c.instance), o(syncProjectKind, c.instance, ordered.Inf)) {
	var project string
	if err := ordered.Decode(key, nil, nil, &project); err != nil {
	c.db.Panic("gerrit client sync decode", "key", storage.Fmt(key), "err", err)
	}
	if err := c.SyncProject(ctx, project); err != nil {
	errs = append(errs, err)
	}
	}
	return errors.Join(errs...)
	}

	// SyncProject syncs a single project.
	func (c *Client) SyncProject(ctx context.Context, project string) (err error) {
	c.slog.Debug("gerrit.SyncProject", "project", project)
	defer func() {
	if err != nil {
	err = fmt.Errorf("SyncProject(%q): %w", project, err)
	}
	}()

	key := o(syncProjectKind, c.instance, project)
	skey := string(key)

	// Lock the project, so that no else is sync'ing concurrently.
	c.db.Lock(skey)
	defer c.db.Unlock(skey)

	// Load sync state.
	var proj projectSync
	if val, ok := c.db.Get(key); !ok {
	return errors.New("missing project")
	} else if err := json.Unmarshal(val, &proj); err != nil {
	return err
	}

	return c.syncChanges(ctx, &proj)
	}

	// syncChanges finds all changes that have been updated since the last
	// sync time. It stores the data for those changes in the database.
	// It also adds in the metadata changes.
	func (c Client) syncChanges(ctx context.Context, proj projectSync) (err error) {
	b := c.db.Batch()
	defer b.Apply()

	// When we need to fetch multiple lists of changes,
	// concurrent modifications can cause us to see the
	// same change more than once. Keep track of the changes
	// we've already seen.
	seen := make(map[int]json.RawMessage)

	// We see the changes from newest to oldest.
	// Update the sync time if we get through
	// all of them with no error.
	lastUpdate := ""
	defer func() {
	if err == nil && lastUpdate != "" {
	proj.SyncDate = lastUpdate
	proj.store(c.db)
	}
	}()

	for change, err := range c.changes(ctx, proj.Name, proj.SyncDate) {
	if err != nil {
	return err
	}
	if err := ctx.Err(); err != nil {
	return err
	}

	if c.flushRequested.Load() {
	// Flush database.
	b.Apply()
	c.db.Flush()
	c.flushRequested.Store(false)
	}

	if lastUpdate == "" {
	// Save the update time of the most recent
	// change to record in the project sync.
	var updated struct {
	Updated string `json:"updated"`
	}
	if err := json.Unmarshal(change, &updated); err != nil {
	return err
	}
	lastUpdate = updated.Updated
	}

	// Change is a Gerrit ChangeInfo in JSON form.
	// Pull out the change number.
	var num struct {
	Number int `json:"_number"`
	MetaID string `json:"meta_rev_id"`
	}
	if err := json.Unmarshal(change, &num); err != nil {
	return err
	}
	changeNum := num.Number
	metaID := num.MetaID
	if changeNum == 0 {
	return fmt.Errorf("missing _number field in %q", change)
	}
	if metaID == "" {
	return fmt.Errorf("missing meta_rev_id field in change %d: %q", changeNum, change)
	}

	if oldChange, ok := seen[changeNum]; ok {
	same, err := sameChangeInfo(change, metaID, oldChange)
	if err != nil {
	return err
	}
	if same {
	// Nothing has changed.
	continue
	}
	}

	seen[changeNum] = change

	key := o(changeKind, c.instance, proj.Name, changeNum)
	if oldChange, ok := c.db.Get(key); ok {
	same, err := sameChangeInfo(change, metaID, oldChange)
	if err != nil {
	return err
	}
	if same {
	// Nothing has changed.
	continue
	}
	}

	b.Set(key, change)
	if err := c.syncComments(ctx, b, proj.Name, changeNum); err != nil {
	return err
	}

	// Record that the change was updated.
	timed.Set(c.db, b, changeUpdateKind, o(c.instance, changeNum, metaID), nil)

	b.MaybeApply()
	}

	return nil
	}

	// syncComments updates the comments of a change in the database.
	func (c *Client) syncComments(ctx context.Context, b storage.Batch, project string, changeNum int) error {
	url := "https://" + c.instance + "/changes/" + strconv.Itoa(changeNum) + "/comments"
	var obj json.RawMessage
	if err := c.get(ctx, url, &obj); err != nil {
	return err
	}

	key := o(commentKind, c.instance, project, changeNum)
	b.Set(key, obj)
	return nil
	}

	// testBefore exists for testing purposes, to only pull changes before
	// some time. For ordinary use this should be the empty string.
	var testBefore string

	// changes returns an iterator over changes in the Gerrit repo.
	// If date is not the empty string, changes only returns changes
	// that were updated after date.
	func (c *Client) changes(ctx context.Context, project, date string) iter.Seq2[json.RawMessage, error] {
	return func(yield func(json.RawMessage, error) bool) {
	baseURL := "https://" + c.instance + "/changes"

	// Gerrit returns up to 500 changes at a time.
	// Gerrit provides a way to skip leading changes,
	// by passing an S parameter.
	// Unfortunately, S only works up to a value of 10,000.
	// Beyond that Gerrit returns no changes.
	// So to get all the changes we have to ask for changes
	// before some time.
	before := testBefore

	for {
	values := url.Values{
	"o": changeOpts,
	}
	query := "p:" + project
	if date != "" {
	query += " after:" + date
	}
	if before != "" {
	query += " before:" + before
	}
	values.Set("q", query)
	addr := baseURL + "?" + values.Encode()

	var body []json.RawMessage
	if err := c.get(ctx, addr, &body); err != nil {
	yield(nil, err)
	return
	}

	if len(body) == 0 {
	return
	}

	// If the last element in body has a
	// _more_changes field, then skip it this time,
	// and pick it up next time with a before query.
	var more struct {
	MoreChanges bool `json:"_more_changes"`
	Updated string `json:"updated"`
	}
	last := body[len(body)-1]
	if err := json.Unmarshal(last, &more); err != nil {
	yield(nil, err)
	return
	}
	if more.MoreChanges {
	if len(body) == 1 {
	yield(nil, errors.New("single change has _more_changes set"))
	}
	body = body[:len(body)-1]
	}

	for _, change := range body {
	if !yield(change, nil) {
	return
	}
	}

	if !more.MoreChanges {
	return
	}

	// Fortunately the before query is inclusive.
	before = more.Updated
	}
	}
	}

	// sameChangeInfo reports whether two ChangeInfo structures,
	// in JSON form, are the same. aMetaID is the meta ID of a.
	func sameChangeInfo(a []byte, aMetaID string, b []byte) (bool, error) {
	if bytes.Equal(a, b) {
	return true, nil
	}

	// Unfortunately Gerrit does not return identical ChangeInfo
	// information with consistent field ordering.
	// In particular, we've seen that the order of "reviewers"
	// can change. So we check the meta ID.
	var extractMetaID struct {
	MetaID string `json:"meta_rev_id"`
	}

	if err := json.Unmarshal(b, &extractMetaID); err != nil {
	return false, err
	}
	bMetaID := extractMetaID.MetaID
	if bMetaID == "" {
	return false, errors.New("missing meta ID")
	}

	return aMetaID == bMetaID, nil
	}

	// get fetches addr and decodes the body as JSON into obj.
	func (c *Client) get(ctx context.Context, addr string, obj any) error {
	c.slog.Info("gerrit GET", "addr", addr)

	tries := 0
	backoff := 1 * time.Second
	for {
	req, err := http.NewRequestWithContext(ctx, "GET", addr, nil)
	if err != nil {
	return err
	}
	resp, err := c.http.Do(req)
	if err != nil {
	return err
	}

	if resp.StatusCode != http.StatusOK {
	data, err := io.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
	return fmt.Errorf("reading body: %v", err)
	}

	if resp.StatusCode == http.StatusTooManyRequests {
	tries++
	if tries > 20 {
	return errors.New("too many requests")
	}
	c.slog.Info("gerrit too many requests",
	"try", tries,
	"sleep", backoff,
	"body", string(data))
	time.Sleep(backoff)
	backoff = min(backoff2, 1time.Minute)

	continue
	}

	return fmt.Errorf("%s\n%s", resp.Status, data)
	}

	// Skip the XSRF header at the start of the response.
	buf := bufio.NewReader(resp.Body)
	defer resp.Body.Close()
	if _, err := buf.ReadSlice('\n'); err != nil {
	return err
	}

	return json.NewDecoder(buf).Decode(obj)
	}
	}