Async Commit/Merge Infrastructure #9732

Annaseli · 2025-11-27T01:18:48Z

Change Description

This PR introduces the infrastructure required for supporting async commit and merge operations.
Part of the full async commit/merge implementation in the Enterprise PR.

Changes

Added four new API endpoints.
Added an AsyncOperationsHandler interface.
Added an OSS no-op implementation of the AsyncOperationsHandler interface.
Updated the catalog protobuf to include the new async-related operations.
Added a status_code field to the generic Task protobuf and refactored error handling to use it.

Not Included in This PR (will be handled in follow-up PRs)

Integration tests.
Handling of expired operations.
Timeout handling.

Testing Details

Validated using the test_async_endpoints.sh script for sanity check (tested from the Enterprise branch).

# Conflicts: # pkg/api/controller.go

…erve

Isan-Rivkin

Thanks, the feature is exciting!
I didn't do full functional review rather learned from the code your intentions :)

Isan-Rivkin · 2025-11-27T04:21:15Z

pkg/catalog/async_operations.go

+	) (*MergeAsyncStatus, error)
+}
+
+type OSSAsyncOperationsHandler struct{}


nit: Rename to a meaningful name based on functionality rather than not location ( i.e NoopCommitHandler , DummyCommitHandler, AsyncCommitHandler )

Isan-Rivkin · 2025-11-27T04:24:26Z

pkg/catalogerrors/errors.go

@@ -0,0 +1,14 @@
+package catalogerrors


Why do we need a dedicated package for catalog errors?
Note: there are errors declated inside catalog package such as catalog/ErrImportClosed, if there is a reason for such package (?) we should be consistent with location.

I added this because I had cyclic dependency issues, but after talking with Barak, I’m updating the implementation and reverting this change.

Isan-Rivkin · 2025-11-27T04:29:30Z

pkg/apierrors/errors.go

@@ -0,0 +1,135 @@
+package apierrors


Why new package?

Because I had cyclic dependency issues, but after talking with Barak, I’m updating the implementation and reverting this change.

Isan-Rivkin · 2025-11-27T04:38:44Z

pkg/catalog/catalog.go

 				log.WithError(err).WithField("step", step.Name).Errorf("Catalog background task step failed")
 				task.Done = true
-				task.Error = err.Error()
+				apierrors.HandleAPIErrorCallback(ctx, log, nil, nil, err, func(w http.ResponseWriter, r *http.Request, code int, v interface{}) {


This seems like the wrong place to translate error into user facing api error.
It should be at the controller level.

The issue is that we like to map what the controller will be doing at this point for the current clients, status code will determine the exact error instead of checking the error message.

Suggest to have a api error callback function that can be set on the catalog, this way the controller can set it's error handler func as the catalog error handler callback.
It will add some complexity when reading the code - but better reuse and handle future changes in the controller.

answered in a comment below.

Isan-Rivkin · 2025-11-27T04:46:52Z

pkg/catalog/catalog.proto

  google.protobuf.Timestamp updated_at = 3;
  int64 progress = 4;
  string error = 5;
+  string error_code = 6; // error classification code for HTTP status mapping


Why do we need 2 error fields? I find this confusing.
Which error do I need to assert for? What do I expect there? Also the fact that it's just HTTP status code does not preserve information regardless what happened. If all we need is HTTP status code then just translate it at the controller.

answered in a comment below.

Remove manually added comments from generated file. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>

nopcoder · 2025-11-27T16:26:36Z

pkg/catalog/catalog.go

 				log.WithError(err).WithField("step", step.Name).Errorf("Catalog background task step failed")
 				task.Done = true
-				task.Error = err.Error()
+				apierrors.HandleAPIErrorCallback(ctx, log, nil, nil, err, func(w http.ResponseWriter, r *http.Request, code int, v interface{}) {


The issue is that we like to map what the controller will be doing at this point for the current clients, status code will determine the exact error instead of checking the error message.

Suggest to have a api error callback function that can be set on the catalog, this way the controller can set it's error handler func as the catalog error handler callback.
It will add some complexity when reading the code - but better reuse and handle future changes in the controller.

nopcoder · 2025-11-27T16:27:20Z

pkg/catalogerrors/errors.go

Assume that with api callback error these errors will return to their place

nopcoder · 2025-11-27T16:29:04Z

pkg/catalog/catalog.proto

  int64 progress = 4;
-  string error = 5;
+  string error_msg = 5;
+  string error_code = 6; // error classification code for HTTP status mapping


Will go with status code as the value will match http status code that the original api would have return

Status code is an integer

nopcoder · 2025-11-27T16:31:40Z

pkg/catalog/catalog.go

 				task.Done = true
-				task.Error = err.Error()
+				apierrors.HandleAPIErrorCallback(ctx, log, nil, nil, err, func(w http.ResponseWriter, r *http.Request, code int, v interface{}) {
+					task.ErrorCode = fmt.Sprintf("%d", code)


The diff is just in ns, but I still prefer to use strconv.Itoa when it comes to int.

Annaseli · 2025-11-27T17:05:34Z

@Isan-Rivkin

Regarding your comments:

In the sync flow, MergeIntoBranch / Commit call Catalog.Merge / Catalog.Commit, which call the Store.Merge and Store.Commit funcs. Errors like 409, 400, 412 reach the controller as proper error objects, allowing handleAPIError to classify them and return the correct HTTP status code.

But in the async flow, the controller submits a task via AsyncOperations.SubmitCommit and AsyncOperations.SubmitMerge, and the actual merge/commit happens later inside RunBackgroundTaskSteps().
Before my change, this function only stored in the kv the error msg:

task.Error = err.Error()

so the original Error() object (and therefore the correct status code) was lost. The async status endpoints had no way to classify the error.

I used HandleAPIErrorCallback inside RunBackgroundTaskSteps() to classify the error before losing the Error() object, and store: Task.ErrorCode, Task.ErrorMsg (renamed from Task.Error)
so the async status endpoints can return the correct HTTP status code.

Using the existing classification logic caused circular dependencies (catalog → api → catalog), so I moved it into a separate apierrors package.

After discussing with Barak, I’m switching to a callback-based approach, so I’ll revert the move and avoid the circular dependency entirely.

itaigilo

Nicely done @Annaseli !

Most of my comments are either nit or about code clarity,
So blocking only since there are a bunch of them.

But other than than, looking good.

api/swagger.yml

itaigilo · 2025-11-27T17:36:25Z

modules/api/factory/build.go

Why this is not part of the catalog module?

moved it there thanks (also in enterprise)

I think it's not pushed yet...

now it is :)

itaigilo · 2025-11-27T17:38:32Z

pkg/api/controller.go

 	licenseManager license.Manager,
 	icebergSyncer icebergsync.Controller,
 	loginTokenProvider authentication.LoginTokenProvider,
+	asyncOperations catalog.AsyncOperationsHandler,


Nit: asyncOpsHandler catalog.AsyncOperationsHandler?
Plus - placing it right after the actionsHandler makes more sense to me. This way, we start piling things up here.

itaigilo · 2025-11-27T17:41:39Z

pkg/api/controller.go

+		return
+	}
+	ctx := r.Context()
+	c.LogAction(ctx, "create_commit_async", r, repository, branch, "")


Suggested change

c.LogAction(ctx, "create_commit_async", r, repository, branch, "")

c.LogAction(ctx, "commit_async", r, repository, branch, "")

itaigilo · 2025-11-27T17:41:48Z

pkg/api/controller.go

+		return
+	}
+	ctx := r.Context()
+	c.LogAction(ctx, "create_commit_async_status", r, repository, branch, "")


Suggested change

c.LogAction(ctx, "create_commit_async_status", r, repository, branch, "")

c.LogAction(ctx, "commit_async_status", r, repository, branch, "")

itaigilo · 2025-11-27T17:43:59Z

pkg/api/controller.go

+		metadata = body.Metadata.AdditionalProperties
+	}
+
+	taskID, err := c.AsyncOperations.SubmitCommit(ctx, repository, branch, body.Message, user.Committer(), metadata, body.Date, params.SourceMetarange, swag.BoolValue(body.AllowEmpty), graveler.WithForce(swag.BoolValue(body.Force)))


In some places it's called SubmitCommit, in others CreateCommit, or just CommitAsync.
Let's align to one of them (or max two), to avoid confusion.

We had "create_commit" only in
c.LogAction(ctx, "commit", r, repository, branch, "") inside the sync Commit function.
I updated it to c.LogAction(ctx, "commit", …) as well, even though that change wasn’t originally mine.

We don’t have a CreateCommit function, what we do have is CreateCommitAction, for example here:

if !c.authorize(w, r, permissions.Node{ Permission: permissions.Permission{ Action: permissions.CreateCommitAction, Resource: permissions.BranchArn(repository, branch), }, }) { return }

But it’s part of the permissions layer.

There is also CommitCreation, but that’s a Swagger schema used by the sync Commit flow, so I can’t rename it.

We also have CreateCommitRecord func in the controller but it's not my code.

In my async code, the only relevant parts are:

In the controller: CommitAsync

In the AsyncOperationsHandler interface: SubmitCommit and GetCommitStatus

Do you think SubmitCommit should be renamed to CreateCommit for consistency, or should we keep the current naming?

Thanks for explaining the rational -
Putting it this way, I guess the naming you chose is good.

As noted below, I assume you should avoid renaming existing log names,
So better to align with these.

pkg/api/controller.go

pkg/catalog/catalog.proto

itaigilo · 2025-11-28T08:58:41Z

modules/catalog/factory/build.go

 	return nil
 }
+
+func NewAsyncOperationsHandler(_ *catalog.Catalog) catalog.AsyncOperationsHandler {


Nit: our convention for these factory methods is (or should be) Build*.
This way it's easier to spot these place where Enterprise is overriding the OSS.

itaigilo · 2025-11-28T09:00:44Z

pkg/api/controller.go

 	Migrator           Migrator
 	Collector          stats.Collector
 	Actions            actionsHandler
+	AsyncOpsHandler    catalog.AsyncOperationsHandler


Can you make it asyncOpsHandler?

itaigilo · 2025-11-28T09:01:43Z

pkg/api/controller.go

 func (c *Controller) handleAPIErrorCallback(ctx context.Context, w http.ResponseWriter, r *http.Request, err error, cb func(w http.ResponseWriter, r *http.Request, code int, v interface{})) bool {
 	// verify if request canceled even if there is no error, early exit point
-	if httputil.IsRequestCanceled(r) {
+	if r != nil && httputil.IsRequestCanceled(r) {


When r can be nil?

Looks kinda weird that this change is introduced only now...

itaigilo · 2025-11-28T09:03:46Z

pkg/api/controller.go

 	}
 	ctx := r.Context()
-	c.LogAction(ctx, "create_commit", r, repository, branch, "")
+	c.LogAction(ctx, "commit", r, repository, branch, "")


I don't think it's a good idea to change log names for existing operations -
Please check if that's the case.

If it's problematic, then prefer keeping the current log name, and align newer logs according to it.

itaigilo · 2025-11-28T09:19:26Z

pkg/catalog/catalog.go

 	UGCPrepareInterval    time.Duration
 	signingKey            config.SecureString
+	// The API callback (handleAPIErrorCallback from the controller) is used to classify the error inside the RunBackgroundTaskSteps function.
+	APIErrorCB func(ctx context.Context, w http.ResponseWriter, r *http.Request, err error, cb func(w http.ResponseWriter, r *http.Request, code int, v interface{})) bool


Maybe extract func(ctx context.Context, w http.ResponseWriter, r *http.Request, err error, cb func(w http.ResponseWriter, r *http.Request, code int, v interface{})) bool into a type, for cleaner code?

Annaseli added 12 commits November 19, 2025 14:46

first steps in implementation

9c5b395

Merge branch 'master' into feature/async-commit-merge

738149b

# Conflicts: # pkg/api/controller.go

Merge branch 'master' into feature/async-commit-merge

88e7ffa

implemented the rest backend funcs

9ab1c3f

Merge branch 'master' into feature/async-commit-merge

3fd36e6

# Conflicts: # pkg/api/controller.go

fixed build issues

f2c2799

added the new asyncOperations interface as params to controller and s…

69135c7

…erve

Add async commit/merge API changes

c7dc897

Merge branch 'master' into feature/async-commit-merge

f6195f1

test 1

e0cd866

test 2

48c87eb

cleaned code

d785292

Annaseli requested a review from a team November 27, 2025 01:18

Annaseli added the include-changelog PR description should be included in next release changelog label Nov 27, 2025

Isan-Rivkin requested changes Nov 27, 2025

View reviewed changes

Annaseli and others added 4 commits November 27, 2025 13:44

Regenerate catalog.pb.go to fix CI validation

120db87

Remove manually added comments from generated file. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>

renamed funcs and cleaned code

69465d6

cleaned code

c904a9f

fixing pointer

3cab6c1

nopcoder requested changes Nov 27, 2025

View reviewed changes

itaigilo requested changes Nov 27, 2025

View reviewed changes

Annaseli added 2 commits November 27, 2025 22:38

updated the error handling code to use callback

15b67b4

addressed comments

0536953

Annaseli requested review from Isan-Rivkin, itaigilo and nopcoder November 27, 2025 22:41

itaigilo reviewed Nov 28, 2025

View reviewed changes

itaigilo requested a review from N-o-Z November 28, 2025 09:26

	c.LogAction(ctx, "create_commit_async", r, repository, branch, "")
	c.LogAction(ctx, "commit_async", r, repository, branch, "")

	c.LogAction(ctx, "create_commit_async_status", r, repository, branch, "")
	c.LogAction(ctx, "commit_async_status", r, repository, branch, "")

Async Commit/Merge Infrastructure #9732

Are you sure you want to change the base?

Async Commit/Merge Infrastructure #9732

Uh oh!

Conversation

Annaseli commented Nov 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Change Description

Changes

Not Included in This PR (will be handled in follow-up PRs)

Testing Details

Uh oh!

Isan-Rivkin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Annaseli commented Nov 27, 2025

Uh oh!

itaigilo left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Annaseli Nov 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Annaseli commented Nov 27, 2025 •

edited

Loading

Annaseli Nov 27, 2025 •

edited

Loading