Implement MVP mirror support in append lifecycle - #1081
Conversation
6db5fc3 to
300f4a3
Compare
82e7391 to
a3494d5
Compare
a3494d5 to
148d465
Compare
AlCutter
changed the title
| // mirrorCheckpoint takes care of mirroring the given checkpoint with the provided mirror policy. | ||
| // Returns signatures from mirrors, ready to append to the checkpoint, or an error. | ||
| func mirrorCheckpoint(ctx context.Context, cp []byte, cpSize uint64, mirrors WitnessGroup, lr LogReader, httpClient *http.Client, opts MirroringOptions) ([]byte, error) { | ||
| func mirrorCheckpoint(ctx context.Context, gw *m_gateway.Gateway, policy *WitnessGroup, cp []byte, size uint64, failOpen bool) ([]byte, error) { |
There was a problem hiding this comment.
| func mirrorCheckpoint(ctx context.Context, gw *m_gateway.Gateway, policy *WitnessGroup, cp []byte, size uint64, failOpen bool) ([]byte, error) { | |
| func mirrorCheckpoint(ctx context.Context, gw *m_gateway.Gateway, policy *WitnessGroup, cp []byte, size uint64, failOpen bool) ([]byte, error) { | |
| // TODO(al): Add metrics | |
| checkPolicy := func(sigs []byte) bool { | |
| newCP := append(slices.Clone(cp), sigs...) | |
| return policy.Satisfied(newCP) | |
| } | |
| exit := func(sigs []byte) (][byte, error) { | |
| newCP, ok := checkPolicy(sigs) | |
| if ok { | |
| return sigs, nil | |
| } | |
| if failOpen { | |
| slog.WarnContext(ctx, "MirrorGateway: policy not met, failing-open") | |
| return sigs, nil | |
| } | |
| return sigs, fmt.Errorf("MirrorGateway: policy not met") | |
| } | |
| return otel.Trace(ctx, "tessera.mirrorCheckpoint", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) { | |
| sigCh := gw.CosignCheckpoint(ctx, cp, size) | |
| var sigBlock bytes.Buffer | |
| for { | |
| select { | |
| case <-ctx.Done(): | |
| return exit(sigBlock.Bytes()) | |
| case sig, ok := <-sigCh: | |
| if !ok { | |
| return exit(sigBlock.Bytes()) | |
| } | |
| sigBlock.Write(sig) | |
| return checkPolicy(sig) | |
| } | |
| } | |
| }) | |
| } |
There was a problem hiding this comment.
Very nit, but I think you can do something along these lines. (but don't apply the suggestion, you'll get into CLA issues)
| } | ||
| } | ||
|
|
||
| // CheckpointPublisherContext returns a function which should be used to create, sign, and potentially witness a new checkpoint. |
There was a problem hiding this comment.
potentially witness and/or mirror a new checkpoint
| defer cancel() | ||
| var err error | ||
| ms, err = mirrorCheckpoint(ctx, cp, size, o.mirrors, lr, httpClient, o.mirrorOpts) | ||
| ms, err = mirrorCheckpoint(mirrorCtx, gw, &o.mirrors, cp, size, o.mirrorOpts.FailOpen) |
There was a problem hiding this comment.
nit: you could move cp, size before &o.mirrors to match with witnessCheckpoint
| if err != nil { | ||
| return nil, fmt.Errorf("failed to parse mirror URLs: %w", err) | ||
| } | ||
| gw, err := m_gateway.NewGateway(ctx, m_gateway.Options{ |
There was a problem hiding this comment.
Should this be skipped is len(mirrorURLs)==0? I don't think it's required, but it could help readability, and avoid "ghost" metrics further down the line like it was was the case with witnessing. You could even check the length of WitnessEndpoint and only parse URLs if need be.
| WithBundleFetcher(opts.LogReader.ReadEntryBundle). | ||
| WithMirrorCheckpointFetcher(mirrorFetcher.ReadCheckpoint) | ||
|
|
||
| c, err := mirror.NewClient(ctx, mOpts) |
There was a problem hiding this comment.
Can the context be removed from NewClient?
| }() | ||
|
|
||
| if err != nil { | ||
| slog.ErrorContext(ctx, "MirrorGateway: Sync failed", slog.String("url", target.url.String()), slog.Any("error", err)) |
There was a problem hiding this comment.
Should this be an error if we're replacing a goal, which hasn't even been picked up by any worker yet?
| // This replacement is "racy", but the worst that can happen is that the worker has already | ||
| // picked up the old goal and we end up simply queuing the new goal instead of replacing the old one. | ||
| select { | ||
| case oldGoal := <-target.goals: |
There was a problem hiding this comment.
Could this behavior get a slow mirror in a state where it never catches up? Should target.goals have a bit of a buffer to have piecemeal uploads?
| g.targets = append(g.targets, target) | ||
|
|
||
| // Start the worker goroutine. | ||
| go g.runWorker(ctx, target) |
There was a problem hiding this comment.
Should this happen in a separate loop? Otherwise, if for some reason client.NewHTTPFetcher fails, then workers that have already been started will stay around.
| slog.DebugContext(cctx, "MirrorGateway: Syncing mirror", slog.String("url", target.url.String()), slog.Uint64("goal", job.cpSize)) | ||
| rSigs, rErr = target.client.Sync(cctx, job.cp, job.cpSize) | ||
| if rErr != nil { | ||
| // TODO(al): Update the client so we can tell whether an error is permanent or transient, and abandon jobs which will never succeed. |
There was a problem hiding this comment.
In the meantime, should goal include cosignCheckpoint's ctx such that chaseGoal can return if this context is being cancelled?
| } | ||
|
|
||
| // CheckpointPublisher returns a function which should be used to create, sign, and potentially witness a new checkpoint. | ||
| // Deprecated: Use CheckpointPublisherContext. |
There was a problem hiding this comment.
Very nit: should validate throw a warning message, or at least prevent CheckpointPublisher, and CheckpointPublisherContext from both being called?
2 participants
This PR adds a simple MVP implementation of mirroring support in the append lifecycle.
This approach is almost certainly going to change, but this will let us make progress in the meantime.
Towards #945