diff --git a/admin/institutional_storage_quota_control/views.py b/admin/institutional_storage_quota_control/views.py
index 7b028da5986..725d464053d 100644
--- a/admin/institutional_storage_quota_control/views.py
+++ b/admin/institutional_storage_quota_control/views.py
@@ -1,16 +1,19 @@
+import logging
 from django.contrib.auth.mixins import UserPassesTestMixin
 from django.db import connection, transaction, IntegrityError
 from django.db.models import Subquery, OuterRef
 from django.http import Http404
-
 from admin.institutions.views import QuotaUserList
-from osf.models import Institution, OSFUser, UserQuota
+from osf.models import Institution, OSFUser, UserQuota, InstitutionDefaultMaxQuota
 from admin.base import settings
 from addons.osfstorage.models import Region
 from django.views.generic import ListView, View
 from django.shortcuts import redirect
 from admin.rdm.utils import RdmPermissionMixin
 from django.core.urlresolvers import reverse
+from api.base import settings as api_settings
+
+logger = logging.getLogger(__name__)
 
 
 class InstitutionStorageList(RdmPermissionMixin, UserPassesTestMixin, ListView):
@@ -122,6 +125,22 @@ def get_institution(self):
             raise Http404
         return institution
 
+    def get_default_max_quota(self):
+        """Get default max quota for the institution, fallback to DEFAULT_MAX_QUOTA when not found."""
+        try:
+            institution_default_max_quota = InstitutionDefaultMaxQuota.objects.get(
+                institution_id=self.institution_id
+            )
+            return institution_default_max_quota.default_max_quota
+        except InstitutionDefaultMaxQuota.DoesNotExist:
+            return api_settings.DEFAULT_MAX_QUOTA
+
+    def get_context_data(self, **kwargs):
+        """ Add default_max_quota to template context """
+        context = super().get_context_data(**kwargs)
+        context['default_max_quota'] = self.get_default_max_quota()
+        return context
+
 
 class UpdateQuotaUserListByInstitutionStorageID(RdmPermissionMixin, UserPassesTestMixin, View):
     """ Change max quota for an institution's users if that institution is not using NII Storage. """
@@ -163,17 +182,23 @@ def post(self, request, *args, **kwargs):
         min_value, max_value = connection.ops.integer_field_range('PositiveIntegerField')
         if min_value <= max_quota <= max_value:
             # If max quota value is between 0 and 2147483647, update or create used quota for each user in the institution
-            for user in OSFUser.objects.filter(
-                    affiliated_institutions=self.institution_id):
-                try:
-                    with transaction.atomic():
-                        UserQuota.objects.update_or_create(
-                            user=user,
-                            storage_type=UserQuota.CUSTOM_STORAGE,
-                            defaults={'max_quota': max_quota}
-                        )
-                except IntegrityError:
-                    UserQuota.objects.filter(user=user, storage_type=UserQuota.CUSTOM_STORAGE).update(max_quota=max_quota)
+            with transaction.atomic():
+                InstitutionDefaultMaxQuota.objects.update_or_create(
+                    institution_id=self.institution_id,
+                    defaults={'default_max_quota': max_quota}
+                )
+                for user in OSFUser.objects.filter(
+                        affiliated_institutions=self.institution_id):
+                    try:
+                        with transaction.atomic():
+                            UserQuota.objects.update_or_create(
+                                user=user,
+                                storage_type=UserQuota.CUSTOM_STORAGE,
+                                defaults={'max_quota': max_quota}
+                            )
+                    except IntegrityError as e:
+                        logger.warning(u'IntegrityError while updating UserQuota: user={}, storage_type={}, max_quota={}: {}.'.format(user.id, UserQuota.CUSTOM_STORAGE, max_quota, str(e)))
+                        UserQuota.objects.filter(user=user, storage_type=UserQuota.CUSTOM_STORAGE).update(max_quota=max_quota)
         return redirect(
             'institutional_storage_quota_control:institution_user_list',
             institution_id=self.institution_id
diff --git a/admin/quota_recalc/views.py b/admin/quota_recalc/views.py
index 29484c3d559..abeb1bf6537 100644
--- a/admin/quota_recalc/views.py
+++ b/admin/quota_recalc/views.py
@@ -1,12 +1,14 @@
+import logging
 from django.http import JsonResponse
 from django.db import transaction, IntegrityError
 
 from addons.osfstorage.models import Region
-from api.base import settings as api_settings
 from osf.models import OSFUser, UserQuota, Node
 from osf.models.node import set_project_storage_type
 from osf.utils.requests import check_select_for_update
-from website.util.quota import used_quota
+from website.util.quota import get_default_max_quota, used_quota
+
+logger = logging.getLogger(__name__)
 
 
 def calculate_quota(user):
@@ -23,6 +25,7 @@ def calculate_quota(user):
     with transaction.atomic():
         for storage_type in storage_type_list:
             used = used_quota(user._id, storage_type)
+            max_quota = get_default_max_quota(user, storage_type)
             try:
                 if check_select_for_update():
                     user_quota = UserQuota.objects.filter(
@@ -42,10 +45,11 @@ def calculate_quota(user):
                         UserQuota.objects.create(
                             user=user,
                             storage_type=storage_type,
-                            max_quota=api_settings.DEFAULT_MAX_QUOTA,
+                            max_quota=max_quota,
                             used=used,
                         )
-                except IntegrityError:
+                except IntegrityError as e:
+                    logger.warning(u'IntegrityError while creating UserQuota in calculate_quota: user={}, storage_type={}: {}.'.format(user.id, storage_type, str(e)))
                     used = used_quota(user._id, storage_type)
                     if check_select_for_update():
                         user_quota = UserQuota.objects.filter(
diff --git a/admin/rdm_custom_storage_location/utils.py b/admin/rdm_custom_storage_location/utils.py
index df350ed2ae1..8dc098487b7 100644
--- a/admin/rdm_custom_storage_location/utils.py
+++ b/admin/rdm_custom_storage_location/utils.py
@@ -13,6 +13,7 @@
 import os
 import owncloud
 from django.core.exceptions import ValidationError
+from django.db import IntegrityError, transaction
 
 from addons.dropboxbusiness.models import node_post_save as dropboxbusiness_post_save
 from addons.onedrivebusiness.models import node_post_save as onedrivebusiness_post_save
@@ -45,7 +46,8 @@
 from framework.exceptions import HTTPError
 from osf.models import AbstractNode
 from website import settings as osf_settings
-from osf.models import Node, OSFUser, ProjectStorageType, UserQuota
+from osf.models import Node, OSFUser, ProjectStorageType, UserQuota, InstitutionDefaultMaxQuota
+from api.base import settings as api_settings
 from osf.models.external import ExternalAccountTemporary, ExternalAccount
 from osf.utils import external_util
 import datetime
@@ -1387,3 +1389,51 @@ def add_node_settings_to_projects(institution, provider_name):
             onedrivebusiness_post_save(None, project, created=project_has_no_node_settings)
         else:
             node_post_save(None, project, created=project_has_no_node_settings)
+
+def is_institution_using_nii_storage(institution):
+    """Return True if the institution is currently using NII storage (or has no region configured)."""
+    old_region = Region.objects.filter(_id=institution._id).first()
+    return (
+        old_region is None or
+        old_region.waterbutler_settings.get('storage', {}).get('type') == Region.NII_STORAGE
+    )
+
+
+def upsert_user_quota_for_institution(institution, provider_short_name, old_is_nii, old_quota_type=None):
+    """Determine quota migration params and upsert UserQuota for all users affiliated with the institution."""
+    new_quota_type = institution.get_user_quota_type_for_nii_storage()
+    users = OSFUser.objects.filter(affiliated_institutions=institution.id)
+
+    if provider_short_name == 'osfstorage' and not old_is_nii:
+        # 機関ストレージ → NII Storage: reset all users to default quota
+        storage_type = new_quota_type
+        quota_map = {}
+    elif provider_short_name != 'osfstorage' and old_is_nii:
+        # NII Storage → 機関ストレージ: set all users to institution default quota
+        default_max_quota = InstitutionDefaultMaxQuota.objects.filter(institution=institution).values_list('default_max_quota', flat=True).first()
+        fixed_quota = default_max_quota if default_max_quota is not None else api_settings.DEFAULT_MAX_QUOTA
+        storage_type = UserQuota.CUSTOM_STORAGE
+        quota_map = {user.id: fixed_quota for user in users}
+    elif old_quota_type is not None and old_quota_type != new_quota_type and new_quota_type is not None:
+        # NII Storage quota type changed (e.g. no-region → has-region): preserve each user's existing quota
+        storage_type = new_quota_type
+        quota_map = {
+            q.user_id: q.max_quota
+            for q in UserQuota.objects.filter(user__in=users, storage_type=old_quota_type)
+        }
+    else:
+        return
+
+    with transaction.atomic():
+        for user in users:
+            max_quota = quota_map.get(user.id, api_settings.DEFAULT_MAX_QUOTA)
+            try:
+                with transaction.atomic():
+                    UserQuota.objects.update_or_create(
+                        user=user,
+                        storage_type=storage_type,
+                        defaults={'max_quota': max_quota}
+                    )
+            except IntegrityError as e:
+                logger.warning(u'IntegrityError while updating UserQuota: user={}, storage_type={}, max_quota={}: {}.'.format(user.id, storage_type, max_quota, str(e)))
+                UserQuota.objects.filter(user=user, storage_type=storage_type).update(max_quota=max_quota)
diff --git a/admin/rdm_custom_storage_location/views.py b/admin/rdm_custom_storage_location/views.py
index dff08d2b3ec..d91276ee0fa 100644
--- a/admin/rdm_custom_storage_location/views.py
+++ b/admin/rdm_custom_storage_location/views.py
@@ -291,6 +291,9 @@ def post(self, request, *args, **kwargs):
                 'message': 'Storage name is missing.'
             }, status=http_status.HTTP_400_BAD_REQUEST)
 
+        old_is_nii = utils.is_institution_using_nii_storage(institution)
+        old_quota_type = institution.get_user_quota_type_for_nii_storage()
+
         result = None
 
         if provider_short_name == 's3':
@@ -436,6 +439,7 @@ def post(self, request, *args, **kwargs):
             utils.change_allowed_for_institutions(
                 institution, provider_short_name)
             utils.add_node_settings_to_projects(institution, provider_short_name)
+            utils.upsert_user_quota_for_institution(institution, provider_short_name, old_is_nii, old_quota_type)
         return JsonResponse(result[0], status=status)
 
 
diff --git a/admin/templates/institutional_storage_quota_control/user_list_by_institute.html b/admin/templates/institutional_storage_quota_control/user_list_by_institute.html
index 2e2050514b0..3d747299e1c 100644
--- a/admin/templates/institutional_storage_quota_control/user_list_by_institute.html
+++ b/admin/templates/institutional_storage_quota_control/user_list_by_institute.html
@@ -121,7 +121,7 @@ <h3>{% trans "Institutional Storage" %} &gt; {{ institution_name }} </h3>
                 <input type="number" name="maxQuota"
                        id="storageLimit" class="form-control flex-item"
                        style="width:70px; max-width: 115px; margin-left: 2px;"
-                       min="0" max="2147483647" required="" value="100"
+                       min="0" max="2147483647" required="" value="{{ default_max_quota|default_if_none:100 }}"
                        oninvalid="this.setCustomValidity('{% trans 'Please enter Each user Quota for Institutional Storage (GB) within 0~2147483647.' %}')"
                        oninput="validateMaxQuotaInput()"
                        onKeyDown="preventDotInput(event)"
diff --git a/admin_tests/institutional_storage_quota_control/test_views.py b/admin_tests/institutional_storage_quota_control/test_views.py
index e5996209d36..9f55b0ce523 100644
--- a/admin_tests/institutional_storage_quota_control/test_views.py
+++ b/admin_tests/institutional_storage_quota_control/test_views.py
@@ -253,6 +253,121 @@ def test_post__max_quota_too_large(self):
         nt.assert_is_not_none(new_user_quota)
         nt.assert_equal(new_user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
 
+    def test_post__institution_default_max_quota_created(self):
+        """Test that InstitutionDefaultMaxQuota is created/updated on POST"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+
+        new_max_quota = 200
+        region = RegionFactory(_id=self.institution01._id)
+        region.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        region.save()
+
+        request = RequestFactory().post(
+            reverse(self.view_name,
+                    kwargs={'institution_id': self.institution01.id}),
+            {'maxQuota': new_max_quota})
+        request.user = self.superuser
+        response = self.view(request, institution_id=self.institution01.id)
+
+        nt.assert_equal(response.status_code, 302)
+
+        # Verify InstitutionDefaultMaxQuota record was created
+        quota = InstitutionDefaultMaxQuota.objects.filter(
+            institution_id=self.institution01.id
+        ).first()
+        nt.assert_is_not_none(quota)
+        nt.assert_equal(quota.default_max_quota, new_max_quota)
+
+    def test_post__institution_default_max_quota_updated(self):
+        """Test that InstitutionDefaultMaxQuota is updated on subsequent POST"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+
+        # Create initial record
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution01.id,
+            defaults={'default_max_quota': 100}
+        )
+
+        # Update via POST
+        new_max_quota = 250
+        region = RegionFactory(_id=self.institution01._id)
+        region.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        region.save()
+
+        request = RequestFactory().post(
+            reverse(self.view_name,
+                    kwargs={'institution_id': self.institution01.id}),
+            {'maxQuota': new_max_quota})
+        request.user = self.superuser
+        response = self.view(request, institution_id=self.institution01.id)
+
+        nt.assert_equal(response.status_code, 302)
+
+        # Verify InstitutionDefaultMaxQuota record was updated
+        quota = InstitutionDefaultMaxQuota.objects.filter(
+            institution_id=self.institution01.id
+        ).first()
+        nt.assert_is_not_none(quota)
+        nt.assert_equal(quota.default_max_quota, new_max_quota)
+
+        # Verify only one record exists for this institution
+        count = InstitutionDefaultMaxQuota.objects.filter(
+            institution_id=self.institution01.id
+        ).count()
+        nt.assert_equal(count, 1)
+
+    def test_post__integrity_error_handles_fallback_update(self):
+        """Test that IntegrityError in update_or_create triggers fallback update logic"""
+        from unittest import mock
+        from django.db import IntegrityError, transaction
+
+        new_max_quota = 300
+        region = RegionFactory(_id=self.institution01.guid)
+        region.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        region.save()
+
+        # Create initial UserQuota
+        initial_quota = UserQuota.objects.create(
+            user=self.institution01_admin,
+            storage_type=UserQuota.CUSTOM_STORAGE,
+            max_quota=100
+        )
+
+        def raise_integrity_error(*args, **kwargs):
+            transaction.set_rollback(True)
+            raise IntegrityError('Simulated race condition')
+
+        # Mock update_or_create to raise IntegrityError and mark transaction for rollback
+        with mock.patch.object(UserQuota.objects, 'update_or_create') as mock_update:
+            mock_update.side_effect = raise_integrity_error
+
+            request = RequestFactory().post(
+                reverse(
+                    self.view_name,
+                    kwargs={'institution_id': self.institution01.id}
+                ),
+                {'maxQuota': new_max_quota}
+            )
+            request.user = self.institution01_admin
+
+            response = self.view(
+                request,
+                institution_id=self.institution01.id
+            )
+
+        # Verify response is redirect
+        nt.assert_equal(response.status_code, 302)
+
+        # Verify max_quota was updated via fallback logic
+        updated_quota = UserQuota.objects.filter(
+            user=self.institution01_admin,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        ).first()
+
+        nt.assert_is_not_none(updated_quota)
+        nt.assert_equal(updated_quota.max_quota, new_max_quota)
+        nt.assert_equal(updated_quota.id, initial_quota.id)
+
 
 class TestUserListByInstitutionStorageID(AdminTestCase):
     def setUp(self):
@@ -261,7 +376,11 @@ def setUp(self):
         self.institution02 = InstitutionFactory(name='inst02')
 
         self.region01 = RegionFactory(_id=self.institution01._id, name='Storage 01')
+        self.region01.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        self.region01.save()
         self.region02 = RegionFactory(_id=self.institution02._id, name='Storage 02')
+        self.region02.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        self.region02.save()
 
         self.anon = AnonymousUser()
         self.user = AuthUserFactory(fullname='user')
@@ -441,18 +560,17 @@ def test_get_institution__not_found(self):
             view.get_institution()
 
         # Institution use NII Storage
-        self.institution01._id = ''
-        self.institution01.save()
+        nii_institution = InstitutionFactory(_id='', name='nii_inst')
         request = RequestFactory().get(
             reverse(
                 self.view_name,
-                kwargs={'institution_id': self.institution01.id}
+                kwargs={'institution_id': nii_institution.id}
             )
         )
         request.user = self.superuser
 
         view = setup_view(self.view_instance, request,
-                          institution_id=self.institution01.id)
+                          institution_id=nii_institution.id)
         with nt.assert_raises(Http404):
             view.get_institution()
 
@@ -476,6 +594,95 @@ def test__institution_id_not_valid(self):
                 )
             )
 
+    def test_get_default_max_quota__institution_has_default(self):
+        """Test get_default_max_quota returns value from DB when record exists"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+
+        # Create default max quota for institution
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution01.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        request = RequestFactory().get(
+            reverse(
+                self.view_name,
+                kwargs={'institution_id': self.institution01.id}
+            )
+        )
+        request.user = self.institution01_admin
+
+        view = setup_view(self.view_instance, request,
+                          institution_id=self.institution01.id)
+        view.institution_id = self.institution01.id
+        default_max_quota = view.get_default_max_quota()
+
+        nt.assert_equal(default_max_quota, 500)
+
+    def test_get_default_max_quota__institution_no_default(self):
+        """Test get_default_max_quota returns DEFAULT_MAX_QUOTA when record doesn't exist"""
+        request = RequestFactory().get(
+            reverse(
+                self.view_name,
+                kwargs={'institution_id': self.institution01.id}
+            )
+        )
+        request.user = self.institution01_admin
+
+        view = setup_view(self.view_instance, request,
+                          institution_id=self.institution01.id)
+        view.institution_id = self.institution01.id
+        default_max_quota = view.get_default_max_quota()
+
+        nt.assert_equal(default_max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_get_context_data__includes_default_max_quota(self):
+        """Test get_context_data includes default_max_quota in context"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+
+        # Create default max quota for institution
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution01.id,
+            defaults={'default_max_quota': 300}
+        )
+
+        request = RequestFactory().get(
+            reverse(
+                self.view_name,
+                kwargs={'institution_id': self.institution01.id}
+            )
+        )
+        request.user = self.institution01_admin
+
+        view = setup_view(self.view_instance, request,
+                          institution_id=self.institution01.id)
+        view.institution_id = self.institution01.id
+        view.object_list = view.get_queryset()
+        context = view.get_context_data()
+
+        nt.assert_in('default_max_quota', context)
+        nt.assert_equal(context['default_max_quota'], 300)
+
+    def test_get_context_data__default_max_quota_fallback(self):
+        """Test get_context_data uses DEFAULT_MAX_QUOTA as fallback"""
+        request = RequestFactory().get(
+            reverse(
+                self.view_name,
+                kwargs={'institution_id': self.institution01.id}
+            )
+        )
+        request.user = self.institution01_admin
+
+        view = setup_view(self.view_instance, request,
+                          institution_id=self.institution01.id)
+        view.institution_id = self.institution01.id
+        view.object_list = view.get_queryset()
+        context = view.get_context_data()
+
+        nt.assert_in('default_max_quota', context)
+        nt.assert_equal(context['default_max_quota'], api_settings.DEFAULT_MAX_QUOTA)
+
+
 class TestAccessInstitutionStorageList(AdminTestCase):
     def setUp(self):
         super(TestAccessInstitutionStorageList, self).setUp()
diff --git a/admin_tests/quota_recalc/test_views.py b/admin_tests/quota_recalc/test_views.py
index cbaa876e015..03c1dc9e862 100644
--- a/admin_tests/quota_recalc/test_views.py
+++ b/admin_tests/quota_recalc/test_views.py
@@ -16,9 +16,11 @@ class TestQuotaRecalcView(AdminTestCase):
     def get_request(view, **kwargs):
         return view(RequestFactory().get('/fake_path'), **kwargs)
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_user_create_userquota_record(self, mock_usedquota):
+    def test_user_create_userquota_record(self, mock_usedquota, mock_get_default_max_quota):
         mock_usedquota.return_value = 1500
+        mock_get_default_max_quota.return_value = api_settings.DEFAULT_MAX_QUOTA
 
         user = AuthUserFactory()
         UserQuota.objects.filter(user=user).delete()
@@ -31,9 +33,11 @@ def test_user_create_userquota_record(self, mock_usedquota):
         nt.assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
         nt.assert_equal(user_quota.used, 1500)
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_user_update_userquota_record(self, mock_usedquota):
+    def test_user_update_userquota_record(self, mock_usedquota, mock_get_default_max_quota):
         mock_usedquota.return_value = 7000
+        mock_get_default_max_quota.return_value = 200
 
         user = AuthUserFactory()
         UserQuota.objects.create(
@@ -51,9 +55,11 @@ def test_user_update_userquota_record(self, mock_usedquota):
         nt.assert_equal(user_quota.max_quota, 200)
         nt.assert_equal(user_quota.used, 7000)
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_user_invalid_guid(self, mock_usedquota):
+    def test_user_invalid_guid(self, mock_usedquota, mock_get_default_max_quota):
         mock_usedquota.return_value = 3000
+        mock_get_default_max_quota.return_value = api_settings.DEFAULT_MAX_QUOTA
 
         response = self.get_request(views.user, guid='cuzidontcare')
         res_json = json.loads(response.content)
@@ -61,9 +67,11 @@ def test_user_invalid_guid(self, mock_usedquota):
         nt.assert_equal(res_json['status'], 'failed')
         nt.assert_equal(res_json['message'], 'User not found.')
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_users_create_userquota_record(self, mock_usedquota):
+    def test_users_create_userquota_record(self, mock_usedquota, mock_get_default_max_quota):
         mock_usedquota.return_value = 1500
+        mock_get_default_max_quota.return_value = api_settings.DEFAULT_MAX_QUOTA
         user = AuthUserFactory()
         user2 = AuthUserFactory()
         UserQuota.objects.filter(user=user).delete()
@@ -94,19 +102,24 @@ def setUp(self):
         super(TestCalculateQuota, self).setUp()
         self.user = AuthUserFactory()
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_user_without_institution(self, mock_usedquota):
+    def test_user_without_institution(self, mock_usedquota, mock_get_default_max_quota):
         mock_usedquota.return_value = 5000
+        mock_get_default_max_quota.return_value = 1000
 
         views.calculate_quota(self.user)
 
         user_quota = UserQuota.objects.filter(user=self.user).all()
         nt.assert_equal(len(user_quota), 1)
         nt.assert_equal(user_quota[0].used, 5000)
+        nt.assert_equal(user_quota[0].max_quota, 1000)
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_user_institution_without_custom_storage(self, mock_usedquota):
+    def test_user_institution_without_custom_storage(self, mock_usedquota, mock_get_default_max_quota):
         mock_usedquota.return_value = 6000
+        mock_get_default_max_quota.return_value = 2000
 
         institution = InstitutionFactory()
         self.user.affiliated_institutions.add(institution)
@@ -116,12 +129,17 @@ def test_user_institution_without_custom_storage(self, mock_usedquota):
         user_quota = UserQuota.objects.filter(user=self.user).all()
         nt.assert_equal(len(user_quota), 1)
         nt.assert_equal(user_quota[0].used, 6000)
+        nt.assert_equal(user_quota[0].max_quota, 2000)
 
+    @mock.patch('admin.quota_recalc.views.get_default_max_quota')
     @mock.patch('admin.quota_recalc.views.used_quota')
-    def test_user_institution_with_custom_storage(self, mock_usedquota):
+    def test_user_institution_with_custom_storage(self, mock_usedquota, mock_get_default_max_quota):
+
         mock_usedquota.side_effect = \
             lambda uid, storage_type: 300 if storage_type == UserQuota.NII_STORAGE else 7000
 
+        mock_get_default_max_quota.side_effect = \
+            lambda _, storage_type: 1000 if storage_type == UserQuota.NII_STORAGE else 2000
         institution = InstitutionFactory()
         self.user.affiliated_institutions.add(institution)
         RegionFactory(_id=institution._id)
@@ -136,5 +154,11 @@ def test_user_institution_with_custom_storage(self, mock_usedquota):
             UserQuota.CUSTOM_STORAGE: 7000,
         }
 
-        nt.assert_equal(user_quota[0].used, expected[user_quota[0].storage_type])
-        nt.assert_equal(user_quota[1].used, expected[user_quota[1].storage_type])
+        expected_max = {
+            UserQuota.NII_STORAGE: 1000,
+            UserQuota.CUSTOM_STORAGE: 2000,
+        }
+
+        for uq in user_quota:
+            nt.assert_equal(uq.used, expected[uq.storage_type])
+            nt.assert_equal(uq.max_quota, expected_max[uq.storage_type])
diff --git a/admin_tests/rdm_custom_storage_location/test_utils.py b/admin_tests/rdm_custom_storage_location/test_utils.py
index 9aa9c9a2d79..55ca340482a 100644
--- a/admin_tests/rdm_custom_storage_location/test_utils.py
+++ b/admin_tests/rdm_custom_storage_location/test_utils.py
@@ -19,18 +19,24 @@
     get_ociinstitutions_info,
     get_nextcloudinstitutions_info,
     get_dropboxbusiness_info,
-    get_institutional_storage_information
+    get_institutional_storage_information,
+    is_institution_using_nii_storage,
+    upsert_user_quota_for_institution,
 )
 from mock import patch, MagicMock
 from osf_tests.factories import (
     InstitutionFactory,
     ProjectFactory,
     RegionFactory,
+    UserFactory,
     bulkmount_waterbutler_settings,
     addon_waterbutler_settings,
     AuthUserFactory
 )
 from tests.base import AdminTestCase
+from api.base import settings as api_settings
+from osf.models import UserQuota
+from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
 from rest_framework import status as http_status
 
 
@@ -438,3 +444,292 @@ def test_get_institutional_storage_information_unknown_provider(self):
         )
 
         nt.assert_equal(result, {})
+
+
+@pytest.mark.feature_202210
+class TestIsInstitutionUsingNiiStorage(AdminTestCase):
+
+    def setUp(self):
+        super(TestIsInstitutionUsingNiiStorage, self).setUp()
+        self.institution = InstitutionFactory()
+
+    def test_returns_true_when_no_region_configured(self):
+        """Return True when institution has no region (no storage configured yet)."""
+        result = is_institution_using_nii_storage(self.institution)
+        nt.assert_true(result)
+
+    def test_returns_true_when_region_is_nii_storage(self):
+        """Return True when institution's region is NII_STORAGE type."""
+        from addons.osfstorage.models import Region
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        result = is_institution_using_nii_storage(self.institution)
+        nt.assert_true(result)
+
+    def test_returns_false_when_region_is_custom_storage(self):
+        """Return False when institution's region is custom (non-NII) storage."""
+        from addons.osfstorage.models import Region
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        region.save()
+
+        result = is_institution_using_nii_storage(self.institution)
+        nt.assert_false(result)
+
+
+@pytest.mark.feature_202210
+class TestUpsertUserQuotaForInstitution(AdminTestCase):
+
+    def setUp(self):
+        super(TestUpsertUserQuotaForInstitution, self).setUp()
+        self.institution = InstitutionFactory()
+        self.user = UserFactory()
+        self.user.affiliated_institutions.add(self.institution)
+        self.user.save()
+
+    def test_nii_to_custom_creates_custom_storage_quota_with_default_max(self):
+        """NII → custom storage: upsert CUSTOM_STORAGE UserQuota with DEFAULT_MAX_QUOTA."""
+        result = upsert_user_quota_for_institution(
+            self.institution, provider_short_name='s3', old_is_nii=True
+        )
+        nt.assert_is_none(result)
+
+        user_quota = UserQuota.objects.get(
+            user=self.user, storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        nt.assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_nii_to_custom_uses_institution_default_max_quota(self):
+        """NII → custom storage: upsert CUSTOM_STORAGE UserQuota with InstitutionDefaultMaxQuota value."""
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='s3', old_is_nii=True
+        )
+
+        user_quota = UserQuota.objects.get(
+            user=self.user, storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        nt.assert_equal(user_quota.max_quota, 500)
+
+    def test_nii_to_custom_updates_existing_quota(self):
+        """NII → custom storage: update max_quota when UserQuota already exists."""
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE,
+            max_quota=100,
+            used=50,
+        )
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 300}
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='s3', old_is_nii=True
+        )
+
+        user_quota = UserQuota.objects.get(
+            user=self.user, storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        nt.assert_equal(user_quota.max_quota, 300)
+        nt.assert_equal(user_quota.used, 50)
+
+    def test_custom_to_nii_creates_quota_with_default_max(self):
+        """Custom → NII storage: upsert UserQuota using get_user_quota_type_for_nii_storage."""
+        from addons.osfstorage.models import Region
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='osfstorage', old_is_nii=False
+        )
+
+        user_quota = UserQuota.objects.get(
+            user=self.user, storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        nt.assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_custom_to_nii_resets_existing_quota_to_default(self):
+        """Custom → NII storage: existing CUSTOM_STORAGE quota is overwritten with DEFAULT_MAX_QUOTA."""
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE,
+            max_quota=500,
+            used=30,
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='osfstorage', old_is_nii=False
+        )
+
+        user_quota = UserQuota.objects.get(user=self.user, storage_type=UserQuota.CUSTOM_STORAGE)
+        nt.assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        nt.assert_equal(user_quota.used, 30)
+
+    def test_no_change_when_already_nii_to_nii(self):
+        """NII re-save (2nd+ time): existing CUSTOM_STORAGE quota must NOT be overwritten."""
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE,
+            max_quota=500,
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution,
+            provider_short_name='osfstorage',
+            old_is_nii=True,
+            old_quota_type=UserQuota.CUSTOM_STORAGE,
+        )
+
+        user_quota = UserQuota.objects.get(user=self.user, storage_type=UserQuota.CUSTOM_STORAGE)
+        nt.assert_equal(user_quota.max_quota, 500)
+
+    def test_no_change_when_custom_to_custom(self):
+        """No quota upsert when switching custom → custom (old_is_nii=False)."""
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='s3', old_is_nii=False
+        )
+
+        nt.assert_false(UserQuota.objects.filter(user=self.user).exists())
+
+    def test_nii_quota_type_changed_preserves_existing_user_quota(self):
+        """NII quota type changed (no-region → NII-region): existing quota of each user is preserved."""
+        from addons.osfstorage.models import Region
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=200,
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution,
+            provider_short_name='osfstorage',
+            old_is_nii=True,
+            old_quota_type=UserQuota.NII_STORAGE,
+        )
+
+        # new_quota_type = CUSTOM_STORAGE (institution now has NII region)
+        user_quota = UserQuota.objects.get(user=self.user, storage_type=UserQuota.CUSTOM_STORAGE)
+        nt.assert_equal(user_quota.max_quota, 200)
+
+    def test_nii_quota_type_changed_uses_default_for_users_without_existing_quota(self):
+        """NII quota type changed: users with no existing quota record receive system default."""
+        from addons.osfstorage.models import Region
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        # No existing UserQuota for user
+        upsert_user_quota_for_institution(
+            self.institution,
+            provider_short_name='osfstorage',
+            old_is_nii=True,
+            old_quota_type=UserQuota.NII_STORAGE,
+        )
+
+        user_quota = UserQuota.objects.get(user=self.user, storage_type=UserQuota.CUSTOM_STORAGE)
+        nt.assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_no_change_when_old_and_new_quota_type_are_same(self):
+        """No quota upsert when old_quota_type equals new_quota_type (NII quota type unchanged)."""
+        # No region → new_quota_type = NII_STORAGE, old_quota_type = NII_STORAGE → same → return
+        upsert_user_quota_for_institution(
+            self.institution,
+            provider_short_name='osfstorage',
+            old_is_nii=True,
+            old_quota_type=UserQuota.NII_STORAGE,
+        )
+
+        nt.assert_false(UserQuota.objects.filter(user=self.user).exists())
+
+    def test_nii_quota_type_changed_mixed_users(self):
+        """B3: user with existing quota gets preserved value; user without quota gets DEFAULT."""
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+
+        user2 = UserFactory()
+        user2.affiliated_institutions.add(self.institution)
+        user2.save()
+
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=300,
+        )
+        # user2 has no NII_STORAGE quota
+
+        upsert_user_quota_for_institution(
+            self.institution,
+            provider_short_name='osfstorage',
+            old_is_nii=True,
+            old_quota_type=UserQuota.NII_STORAGE,
+        )
+
+        quota1 = UserQuota.objects.get(user=self.user, storage_type=UserQuota.CUSTOM_STORAGE)
+        quota2 = UserQuota.objects.get(user=user2, storage_type=UserQuota.CUSTOM_STORAGE)
+        nt.assert_equal(quota1.max_quota, 300)
+        nt.assert_equal(quota2.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_non_affiliated_user_quota_not_affected(self):
+        """Users not affiliated with the institution must not have their quota created or modified."""
+        non_affiliated_user = UserFactory()
+        # non_affiliated_user is deliberately NOT added to self.institution
+
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 400}
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='s3', old_is_nii=True
+        )
+
+        nt.assert_false(
+            UserQuota.objects.filter(user=non_affiliated_user).exists()
+        )
+
+    @patch('admin.rdm_custom_storage_location.utils.UserQuota.objects.update_or_create')
+    def test_integrity_error_falls_back_to_filter_update(self, mock_uoc):
+        """On IntegrityError, fall back to filter().update() to set max_quota."""
+        from django.db import IntegrityError
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 200}
+        )
+        mock_uoc.side_effect = IntegrityError('duplicate key')
+
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE,
+            max_quota=50,
+            used=10,
+        )
+
+        upsert_user_quota_for_institution(
+            self.institution, provider_short_name='s3', old_is_nii=True
+        )
+
+        user_quota = UserQuota.objects.get(
+            user=self.user, storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        nt.assert_equal(user_quota.max_quota, 200)
diff --git a/admin_tests/rdm_custom_storage_location/test_views.py b/admin_tests/rdm_custom_storage_location/test_views.py
index fac5695ae67..6385d9cd5bf 100644
--- a/admin_tests/rdm_custom_storage_location/test_views.py
+++ b/admin_tests/rdm_custom_storage_location/test_views.py
@@ -17,6 +17,8 @@
     RegionFactory,
     InstitutionFactory,
 )
+from rest_framework import status as http_status
+from osf.models import UserQuota
 
 
 class TestInstitutionDefaultStorage(AdminTestCase):
@@ -115,7 +117,19 @@ def test_get_without_custom_storage(self, *args, **kwargs):
         nt.assert_equal(res.context_data['selected_provider_short_name'], 'osfstorage')
 
     def test_get_custom_storage(self, *args, **kwargs):
-        self.us = RegionFactory()
+        custom_settings = {
+            'storage': {
+                'provider': 'glowcloud',
+                'container': 'osf_storage',
+                'use_public': True,
+                'bucket': 'bucket test',
+                'folder': {'encrypt_uploads': 'encrypt upload test'},
+            },
+            'extended': {'base_folder': 'base folder'},
+            'admin_dbmid': 'abc',
+            'team_folder_id': '1',
+        }
+        self.us = RegionFactory(waterbutler_settings=custom_settings)
         self.us._id = self.institution1._id
         self.us.save()
         res = self.view.get(self.request, *args, **kwargs)
@@ -720,3 +734,103 @@ def test_get_superuser_institution_not_exist(self):
         response = self.view_get({'provider': 'test'})
         self.assertEquals(response.status_code, 404)
         self.assertEquals(response.content, b'{"message": "Institution does not exist"}')
+
+
+class TestSaveCredentialsViewUpsertUserQuota(AdminTestCase):
+    """Test that SaveCredentialsView calls upsert_user_quota_for_institution after a successful save."""
+
+    def setUp(self):
+        super(TestSaveCredentialsViewUpsertUserQuota, self).setUp()
+        self.user = AuthUserFactory()
+        self.user.is_staff = True
+        self.user.is_superuser = True
+        self.user.save()
+        self.institution = InstitutionFactory()
+        self.institution_id = self.institution.id
+
+    def _post(self, params):
+        from django.test import RequestFactory
+        request = RequestFactory().post(
+            'fake_path',
+            json.dumps(params),
+            content_type='application/json'
+        )
+        request.user = self.user
+        return views.SaveCredentialsView.as_view()(request, institution_id=self.institution_id)
+
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.upsert_user_quota_for_institution')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.add_node_settings_to_projects')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.change_allowed_for_institutions')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.update_nodes_storage')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.save_osfstorage_credentials')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.is_institution_using_nii_storage')
+    def test_upsert_user_quota_called_on_success(
+        self, mock_is_nii, mock_save, mock_update_nodes, mock_change_allowed,
+        mock_add_node, mock_upsert
+    ):
+        """upsert_user_quota_for_institution is called with correct args when save succeeds."""
+        mock_is_nii.return_value = False
+        mock_save.return_value = ({'message': 'NII storage was set successfully'}, http_status.HTTP_200_OK)
+
+        self._post({'provider_short_name': 'osfstorage'})
+
+        # old_quota_type is UserQuota.NII_STORAGE because institution has no region configured
+        mock_upsert.assert_called_once_with(self.institution, 'osfstorage', False, UserQuota.NII_STORAGE)
+
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.upsert_user_quota_for_institution')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.save_osfstorage_credentials')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.is_institution_using_nii_storage')
+    def test_upsert_user_quota_not_called_on_failure(
+        self, mock_is_nii, mock_save, mock_upsert
+    ):
+        """upsert_user_quota_for_institution is NOT called when save returns error."""
+        mock_is_nii.return_value = True
+        mock_save.return_value = ({'message': 'Some error'}, http_status.HTTP_400_BAD_REQUEST)
+
+        self._post({'provider_short_name': 'osfstorage'})
+
+        mock_upsert.assert_not_called()
+
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.upsert_user_quota_for_institution')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.add_node_settings_to_projects')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.change_allowed_for_institutions')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.update_nodes_storage')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.save_osfstorage_credentials')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.is_institution_using_nii_storage')
+    def test_old_is_nii_captured_before_save(
+        self, mock_is_nii, mock_save, mock_update_nodes, mock_change_allowed,
+        mock_add_node, mock_upsert
+    ):
+        """old_is_nii value is captured before credentials are saved and passed to upsert."""
+        mock_is_nii.return_value = True
+        mock_save.return_value = ({'message': 'NII storage was set successfully'}, http_status.HTTP_200_OK)
+
+        self._post({'provider_short_name': 'osfstorage'})
+
+        # old_quota_type is UserQuota.NII_STORAGE because institution has no region configured
+        mock_upsert.assert_called_once_with(self.institution, 'osfstorage', True, UserQuota.NII_STORAGE)
+
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.upsert_user_quota_for_institution')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.add_node_settings_to_projects')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.change_allowed_for_institutions')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.update_nodes_storage')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.save_osfstorage_credentials')
+    @mock.patch('admin.rdm_custom_storage_location.views.utils.is_institution_using_nii_storage')
+    def test_old_quota_type_captured_before_save(
+        self, mock_is_nii, mock_save, mock_update_nodes, mock_change_allowed,
+        mock_add_node, mock_upsert
+    ):
+        """old_quota_type is captured before credentials are saved and passed to upsert."""
+        from osf_tests.factories import RegionFactory
+        from addons.osfstorage.models import Region as OsfRegion
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = OsfRegion.NII_STORAGE
+        region.save()
+
+        mock_is_nii.return_value = True
+        mock_save.return_value = ({'message': 'NII storage was set successfully'}, http_status.HTTP_200_OK)
+
+        self._post({'provider_short_name': 'osfstorage'})
+
+        # Institution has a NII region → old_quota_type = UserQuota.CUSTOM_STORAGE
+        mock_upsert.assert_called_once_with(self.institution, 'osfstorage', True, UserQuota.CUSTOM_STORAGE)
diff --git a/osf/migrations/0272_institutiondefaultmaxquota.py b/osf/migrations/0272_institutiondefaultmaxquota.py
new file mode 100644
index 00000000000..92a4cc68639
--- /dev/null
+++ b/osf/migrations/0272_institutiondefaultmaxquota.py
@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.28 on 2026-07-23 16:23
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+import django.db.models.deletion
+import django_extensions.db.fields
+import osf.models.base
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('osf', '0271_merge_20260622_1735'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='InstitutionDefaultMaxQuota',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created', django_extensions.db.fields.CreationDateTimeField(auto_now_add=True, verbose_name='created')),
+                ('modified', django_extensions.db.fields.ModificationDateTimeField(auto_now=True, verbose_name='modified')),
+                ('default_max_quota', models.IntegerField(default=100)),
+                ('institution', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='default_max_quota', to='osf.Institution')),
+            ],
+            options={
+                'db_table': 'osf_institution_default_max_quota',
+            },
+            bases=(models.Model, osf.models.base.QuerySetExplainMixin),
+        ),
+    ]
diff --git a/osf/models/__init__.py b/osf/models/__init__.py
index c86fc3973ad..1a66afdadf1 100644
--- a/osf/models/__init__.py
+++ b/osf/models/__init__.py
@@ -64,6 +64,7 @@
 from osf.models.project_storage_type import ProjectStorageType  # noqa
 from osf.models.region_external_account import RegionExternalAccount  # noqa
 from osf.models.institution_entitlement import InstitutionEntitlement  # noqa
+from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota  # noqa
 from osf.models.export_data_location import ExportDataLocation  # noqa
 from osf.models.export_data import ExportData  # noqa
 from osf.models.export_data_restore import ExportDataRestore  # noqa
diff --git a/osf/models/institution_default_max_quota.py b/osf/models/institution_default_max_quota.py
new file mode 100644
index 00000000000..8b18e9f7c1c
--- /dev/null
+++ b/osf/models/institution_default_max_quota.py
@@ -0,0 +1,29 @@
+from django.db import models
+
+from osf.models import Institution
+from osf.models.base import BaseModel
+
+
+class InstitutionDefaultMaxQuota(BaseModel):
+    institution = models.OneToOneField(Institution, related_name='default_max_quota', on_delete=models.CASCADE)
+    default_max_quota = models.IntegerField(default=100)
+
+    class Meta:
+        db_table = 'osf_institution_default_max_quota'
+
+    @classmethod
+    def get_quota_by_user(cls, user_id):
+        """
+        Get the default maximum quota for the user's affiliated institution.
+
+        Args:
+            user_id (int): The ID of the user whose affiliated institution's default quota is to be retrieved.
+
+        Returns:
+            int | None: The default maximum quota (in GB) for the user's affiliated institution.
+                Returns None if the user is not affiliated with an institution or if no default quota is configured.
+        """
+        return cls.objects.filter(
+            institution__osfuser__id=user_id,
+            institution__is_deleted=False
+        ).extra(order_by=['osf_osfuser_affiliated_institutions.id']).values_list('default_max_quota', flat=True).first()
diff --git a/tests/test_institution_default_max_quota.py b/tests/test_institution_default_max_quota.py
new file mode 100644
index 00000000000..af02b5a0d63
--- /dev/null
+++ b/tests/test_institution_default_max_quota.py
@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+from nose import tools as nt
+
+from osf.models import InstitutionDefaultMaxQuota
+from osf_tests.factories import AuthUserFactory, InstitutionFactory
+from tests.base import OsfTestCase
+
+
+class TestInstitutionDefaultMaxQuota(OsfTestCase):
+
+    def test_get_quota_by_user_has_institution(self):
+        user = AuthUserFactory()
+        institution = InstitutionFactory()
+
+        user.affiliated_institutions.add(institution)
+
+        InstitutionDefaultMaxQuota.objects.create(
+            institution=institution,
+            default_max_quota=500
+        )
+
+        result = InstitutionDefaultMaxQuota.get_quota_by_user(user.id)
+
+        nt.assert_equal(result, 500)
+
+    def test_get_quota_by_user_no_institution(self):
+        user = AuthUserFactory()
+
+        result = InstitutionDefaultMaxQuota.get_quota_by_user(user.id)
+
+        nt.assert_equal(result, None)
+
+    def test_get_quota_by_user_multiple_institutions(self):
+        user = AuthUserFactory()
+
+        inst1 = InstitutionFactory()
+        inst2 = InstitutionFactory()
+
+        user.affiliated_institutions.add(inst1)
+        user.affiliated_institutions.add(inst2)
+
+        InstitutionDefaultMaxQuota.objects.create(
+            institution=inst1,
+            default_max_quota=100
+        )
+
+        InstitutionDefaultMaxQuota.objects.create(
+            institution=inst2,
+            default_max_quota=200
+        )
+
+        result = InstitutionDefaultMaxQuota.get_quota_by_user(user.id)
+
+        nt.assert_equal(result, 100)
diff --git a/tests/test_quota.py b/tests/test_quota.py
index 2fc183213d0..4c324b0c5db 100644
--- a/tests/test_quota.py
+++ b/tests/test_quota.py
@@ -4,7 +4,7 @@
 from nose.tools import *  # noqa (PEP8 asserts)
 import pytest
 
-from addons.osfstorage.models import OsfStorageFileNode
+from addons.osfstorage.models import OsfStorageFileNode, Region
 from api.base import settings as api_settings
 from framework.auth import signing
 from tests.base import OsfTestCase
@@ -16,6 +16,7 @@
 )
 from osf.utils.requests import check_select_for_update
 from website.util import web_url_for, quota
+from django.db import IntegrityError
 
 
 @pytest.mark.enable_implicit_clean
@@ -1532,6 +1533,36 @@ def test_update_user_used_quota_method_for_recalculate_quota_process__nii_custom
         user_quota = user_quota.first()
         assert_equal(user_quota.used, 1000)
 
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    @mock.patch('website.util.quota.used_quota')
+    def test_update_user_used_quota_without_select_for_update_existing(self, mock_used, _mock_check):
+        """update_user_used_quota should work without select_for_update when UserQuota exists"""
+        mock_used.return_value = 750
+
+        quota.update_user_used_quota(
+            user=self.user,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.user, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.used, 750)
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    @mock.patch('website.util.quota.used_quota')
+    def test_update_user_used_quota_without_select_for_update_create(self, mock_used, _mock_check):
+        """update_user_used_quota should work without select_for_update when creating new UserQuota"""
+        another_user = UserFactory()
+        mock_used.return_value = 600
+
+        quota.update_user_used_quota(
+            user=another_user,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=another_user, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.used, 600)
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
 
 class TestQuotaApiWaterbutler(OsfTestCase):
     def setUp(self):
@@ -1661,3 +1692,755 @@ def test_used_half_custom_institution_quota(self):
         assert_equal(response.status_code, 200)
         assert_equal(response.json['max'], 200 * api_settings.SIZE_UNIT_GB)
         assert_equal(response.json['used'], 100 * api_settings.SIZE_UNIT_GB)
+
+
+@pytest.mark.enable_implicit_clean
+class TestGetQuotaInfoWithInstitutionDefaultMaxQuota(OsfTestCase):
+    """Test get_quota_info() with InstitutionDefaultMaxQuota logic"""
+
+    def setUp(self):
+        super(TestGetQuotaInfoWithInstitutionDefaultMaxQuota, self).setUp()
+        self.user = AuthUserFactory()
+        self.institution = InstitutionFactory(name='Test Institution')
+        self.user.affiliated_institutions.add(self.institution)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_get_quota_info_nii_storage_uses_default(self, mock_used_quota):
+        """NII_STORAGE should always use DEFAULT_MAX_QUOTA, ignoring InstitutionDefaultMaxQuota"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 50
+
+        # Set institution quota
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        # NII_STORAGE should ignore institution quota
+        max_quota, used = quota.get_quota_info(self.user, UserQuota.NII_STORAGE)
+
+        assert_equal(max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(used, 50)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_get_quota_info_custom_storage_with_institution_quota(self, mock_used_quota):
+        """CUSTOM_STORAGE should use InstitutionDefaultMaxQuota when available"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 100
+
+        # Set institution quota
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 350}
+        )
+
+        # CUSTOM_STORAGE should use institution quota
+        max_quota, used = quota.get_quota_info(self.user, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(max_quota, 350)
+        assert_equal(used, 100)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_get_quota_info_custom_storage_fallback_to_default(self, mock_used_quota):
+        """CUSTOM_STORAGE should fallback to DEFAULT_MAX_QUOTA when no InstitutionDefaultMaxQuota"""
+        mock_used_quota.return_value = 75
+
+        # No InstitutionDefaultMaxQuota set
+        max_quota, used = quota.get_quota_info(self.user, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(used, 75)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_get_quota_info_with_existing_userquota(self, mock_used_quota):
+        """get_quota_info should return existing UserQuota, not use InstitutionDefaultMaxQuota"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 0
+
+        # Create UserQuota with custom max_quota
+        UserQuota.objects.create(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE,
+            max_quota=600,
+            used=200
+        )
+
+        # Set institution quota (should be ignored)
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 350}
+        )
+
+        # Should return existing UserQuota, not institution quota
+        max_quota, used = quota.get_quota_info(self.user, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(max_quota, 600)
+        assert_equal(used, 200)
+
+
+@pytest.mark.enable_implicit_clean
+class TestFileAddedWithInstitutionDefaultMaxQuota(OsfTestCase):
+    """Test file_added() with InstitutionDefaultMaxQuota logic"""
+
+    def setUp(self):
+        super(TestFileAddedWithInstitutionDefaultMaxQuota, self).setUp()
+        self.user = AuthUserFactory()
+        self.institution = InstitutionFactory(name='Test Institution')
+        self.region = RegionFactory(_id=self.institution._id)
+        self.region.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        self.region.save()
+        self.user.affiliated_institutions.add(self.institution)
+        self.project = ProjectFactory(creator=self.user, is_deleted=False)
+        self.file_node = mock.MagicMock()
+
+    def tearDown(self):
+        super(TestFileAddedWithInstitutionDefaultMaxQuota, self).tearDown()
+        if hasattr(self, 'project') and self.project:
+            self.project.delete()
+
+    @mock.patch('osf.models.FileInfo.objects.create')
+    @mock.patch('website.util.quota.used_quota')
+    def test_file_added_custom_storage_with_institution_quota(self, mock_used_quota, mock_create):
+        """file_added should create UserQuota with InstitutionDefaultMaxQuota value"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 0
+
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 400}
+        )
+
+        payload = {'metadata': {'size': '100'}}
+        quota.file_added(self.project, payload, self.file_node, UserQuota.CUSTOM_STORAGE)
+
+        user_quota = UserQuota.objects.get(
+            user=self.project.creator,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        assert_equal(user_quota.max_quota, 400)
+        assert_equal(user_quota.used, 100)
+
+    @mock.patch('osf.models.FileInfo.objects.create')
+    @mock.patch('website.util.quota.used_quota')
+    def test_file_added_custom_storage_without_institution_quota(self, mock_used_quota, mock_create):
+        """file_added should fallback to DEFAULT_MAX_QUOTA when no InstitutionDefaultMaxQuota"""
+        mock_used_quota.return_value = 0
+
+        # No InstitutionDefaultMaxQuota
+        payload = {'metadata': {'size': '150'}}
+        quota.file_added(self.project, payload, self.file_node, UserQuota.CUSTOM_STORAGE)
+
+        user_quota = UserQuota.objects.get(
+            user=self.project.creator,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 150)
+
+    @mock.patch('osf.models.FileInfo.objects.create')
+    @mock.patch('website.util.quota.used_quota')
+    def test_file_added_nii_storage_ignores_institution_quota(self, mock_used_quota, mock_create):
+        """file_added should use DEFAULT_MAX_QUOTA for NII_STORAGE, ignoring InstitutionDefaultMaxQuota"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 0
+
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 400}
+        )
+
+        payload = {'metadata': {'size': '75'}}
+        quota.file_added(self.project, payload, self.file_node, UserQuota.NII_STORAGE)
+
+        user_quota = UserQuota.objects.get(
+            user=self.project.creator,
+            storage_type=UserQuota.NII_STORAGE
+        )
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 75)
+
+
+@pytest.mark.enable_implicit_clean
+class TestUpdateUserUsedQuotaWithInstitutionDefaultMaxQuota(OsfTestCase):
+    """Test update_user_used_quota() with InstitutionDefaultMaxQuota logic"""
+
+    def setUp(self):
+        super(TestUpdateUserUsedQuotaWithInstitutionDefaultMaxQuota, self).setUp()
+        self.user = UserFactory()
+        self.user.save()
+        self.institution = InstitutionFactory(name='Test Institution')
+        self.user.affiliated_institutions.add(self.institution)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_update_user_used_quota_create_custom_storage_with_institution_quota(self, mock_used_quota):
+        """update_user_used_quota should create UserQuota with InstitutionDefaultMaxQuota value"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 300
+
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        quota.update_user_used_quota(self.user, UserQuota.CUSTOM_STORAGE)
+
+        user_quota = UserQuota.objects.get(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        assert_equal(user_quota.max_quota, 500)
+        assert_equal(user_quota.used, 300)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_update_user_used_quota_create_custom_storage_without_institution_quota(self, mock_used_quota):
+        """update_user_used_quota should fallback to DEFAULT_MAX_QUOTA when no InstitutionDefaultMaxQuota"""
+        mock_used_quota.return_value = 250
+
+        # No InstitutionDefaultMaxQuota
+        quota.update_user_used_quota(self.user, UserQuota.CUSTOM_STORAGE)
+
+        user_quota = UserQuota.objects.get(
+            user=self.user,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 250)
+
+    @mock.patch('website.util.quota.used_quota')
+    def test_update_user_used_quota_nii_storage_ignores_institution_quota(self, mock_used_quota):
+        """update_user_used_quota should use DEFAULT_MAX_QUOTA for NII_STORAGE"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        mock_used_quota.return_value = 200
+
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        quota.update_user_used_quota(self.user, UserQuota.NII_STORAGE)
+
+        user_quota = UserQuota.objects.get(
+            user=self.user,
+            storage_type=UserQuota.NII_STORAGE
+        )
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 200)
+
+
+class TestGetDefaultMaxQuota(OsfTestCase):
+    """Unit tests for get_default_max_quota()"""
+
+    def setUp(self):
+        super(TestGetDefaultMaxQuota, self).setUp()
+        self.user = AuthUserFactory()
+        self.institution = InstitutionFactory(name='Test Institution')
+        self.user.affiliated_institutions.add(self.institution)
+
+    def _make_non_nii_region(self):
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.INSTITUTIONS
+        region.save()
+        return region
+
+    def _make_nii_region(self):
+        region = RegionFactory(_id=self.institution._id)
+        region.waterbutler_settings['storage']['type'] = Region.NII_STORAGE
+        region.save()
+        return region
+
+    def test_nii_storage_returns_default(self):
+        """NII_STORAGE always returns DEFAULT_MAX_QUOTA regardless of institution quota"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        self._make_non_nii_region()
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        result = quota.get_default_max_quota(self.user, UserQuota.NII_STORAGE)
+
+        assert_equal(result, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_custom_storage_no_institution_returns_default(self):
+        """CUSTOM_STORAGE returns DEFAULT_MAX_QUOTA when user has no affiliated institution"""
+        user_no_inst = AuthUserFactory()
+
+        result = quota.get_default_max_quota(user_no_inst, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(result, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_custom_storage_institution_uses_nii_storage_returns_default(self):
+        """CUSTOM_STORAGE returns DEFAULT_MAX_QUOTA when institution's region is NII_STORAGE"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        self._make_nii_region()
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 500}
+        )
+
+        result = quota.get_default_max_quota(self.user, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(result, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_custom_storage_no_institution_quota_returns_default(self):
+        """CUSTOM_STORAGE returns DEFAULT_MAX_QUOTA when no InstitutionDefaultMaxQuota is set"""
+        self._make_non_nii_region()
+        # No InstitutionDefaultMaxQuota created
+
+        result = quota.get_default_max_quota(self.user, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(result, api_settings.DEFAULT_MAX_QUOTA)
+
+    def test_custom_storage_with_institution_quota_returns_institution_value(self):
+        """CUSTOM_STORAGE returns institution quota when region is non-NII and quota is set"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+        self._make_non_nii_region()
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=self.institution.id,
+            defaults={'default_max_quota': 350}
+        )
+
+        result = quota.get_default_max_quota(self.user, UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(result, 350)
+
+
+@pytest.mark.enable_implicit_clean
+class TestFileAddedWithoutSelectForUpdate(OsfTestCase):
+    """Test file_added() without select_for_update branch"""
+
+    def setUp(self):
+        super(TestFileAddedWithoutSelectForUpdate, self).setUp()
+        self.user = UserFactory()
+        self.project_creator = UserFactory()
+        self.node = ProjectFactory(creator=self.project_creator)
+        self.file = OsfStorageFileNode.create(
+            target=self.node,
+            path='/testfile',
+            _id='testfile',
+            name='testfile',
+            materialized_path='/testfile'
+        )
+        self.file.save()
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_file_added_without_select_for_update_new_userquota(self, _mock_check):
+        """file_added should work without select_for_update when creating new UserQuota"""
+        assert_false(UserQuota.objects.filter(user=self.project_creator).exists())
+
+        payload = {'metadata': {'size': 1000}}
+        quota.file_added(self.node, payload, self.file, UserQuota.NII_STORAGE)
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.used, 1000)
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_file_added_without_select_for_update_existing_userquota(self, _mock_check):
+        """file_added should work without select_for_update when updating existing UserQuota"""
+        UserQuota.objects.create(
+            user=self.project_creator,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=500,
+            used=2000
+        )
+
+        payload = {'metadata': {'size': 500}}
+        quota.file_added(self.node, payload, self.file, UserQuota.NII_STORAGE)
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.used, 2500)
+        assert_equal(user_quota.max_quota, 500)
+
+
+@pytest.mark.enable_implicit_clean
+class TestNodeRemovedWithoutSelectForUpdate(OsfTestCase):
+    """Test node_removed() without select_for_update branch"""
+
+    def setUp(self):
+        super(TestNodeRemovedWithoutSelectForUpdate, self).setUp()
+        self.user = UserFactory()
+        self.project_creator = UserFactory()
+        self.node = ProjectFactory(creator=self.project_creator)
+        self.file = OsfStorageFileNode.create(
+            target=self.node,
+            path='/testfile',
+            _id='testfile',
+            name='testfile',
+            materialized_path='/testfile'
+        )
+        self.file.save()
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_node_removed_without_select_for_update(self, _mock_check):
+        """node_removed should work without select_for_update when deleting file"""
+        UserQuota.objects.create(
+            user=self.project_creator,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=api_settings.DEFAULT_MAX_QUOTA,
+            used=1500
+        )
+        FileInfo.objects.create(file=self.file, file_size=1000)
+
+        self.file.deleted_on = datetime.datetime.now()
+        self.file.deleted_by = self.user
+        self.file.type = 'osf.trashedfile'
+        self.file.save()
+
+        quota.node_removed(self.node, self.user, {}, self.file, UserQuota.NII_STORAGE)
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.used, 500)
+
+
+@pytest.mark.enable_implicit_clean
+class TestUserQuotaGetOrCreate(OsfTestCase):
+    """Test get_or_create logic in file_modified() function"""
+
+    def setUp(self):
+        super(TestUserQuotaGetOrCreate, self).setUp()
+        self.user = UserFactory()
+        self.project_creator = UserFactory()
+        self.node = ProjectFactory(creator=self.project_creator)
+        self.file = OsfStorageFileNode.create(
+            target=self.node,
+            path='/testfile',
+            _id='testfile',
+            name='testfile',
+            materialized_path='/testfile'
+        )
+        self.file.save()
+
+    def test_get_or_create_creates_new_userquota(self):
+        """get_or_create should create new UserQuota if it doesn't exist"""
+        assert_false(UserQuota.objects.filter(user=self.project_creator).exists())
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 5000
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(len(user_quota), 1)
+        user_quota = user_quota[0]
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 5000)
+
+    def test_get_or_create_retrieves_existing_userquota(self):
+        """get_or_create should retrieve existing UserQuota without creating a new one"""
+        UserQuota.objects.create(
+            user=self.project_creator,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=200,
+            used=100
+        )
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 200
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota_list = UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(len(user_quota_list), 1)
+        user_quota = user_quota_list[0]
+        assert_equal(user_quota.max_quota, 200)
+        assert_equal(user_quota.used, 300)
+
+    def test_get_or_create_with_custom_storage_new_userquota(self):
+        """get_or_create should work correctly with CUSTOM_STORAGE"""
+        institution = InstitutionFactory()
+        self.project_creator.affiliated_institutions.add(institution)
+        RegionFactory(_id=institution._id)
+        ProjectStorageType.objects.filter(node=self.node).update(
+            storage_type=ProjectStorageType.CUSTOM_STORAGE
+        )
+
+        assert_false(UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.CUSTOM_STORAGE).exists())
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 3000
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.CUSTOM_STORAGE)
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 3000)
+
+    def test_get_or_create_with_default_max_quota_logic(self):
+        """get_or_create should use get_default_max_quota() for max_quota value"""
+        from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
+
+        institution = InstitutionFactory()
+        self.project_creator.affiliated_institutions.add(institution)
+        RegionFactory(_id=institution._id)
+        ProjectStorageType.objects.filter(node=self.node).update(
+            storage_type=ProjectStorageType.CUSTOM_STORAGE
+        )
+
+        custom_max = 450
+        InstitutionDefaultMaxQuota.objects.update_or_create(
+            institution_id=institution.id,
+            defaults={'default_max_quota': custom_max}
+        )
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 2000
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.CUSTOM_STORAGE)
+        assert_equal(user_quota.max_quota, custom_max)
+        assert_equal(user_quota.used, 2000)
+
+    def test_get_or_create_multiple_storage_types(self):
+        """get_or_create should handle multiple storage types for same user"""
+        institution = InstitutionFactory()
+        self.project_creator.affiliated_institutions.add(institution)
+        RegionFactory(_id=institution._id)
+
+        # Create separate files for each storage type
+        file_nii = OsfStorageFileNode.create(
+            target=self.node,
+            path='/testfile_nii',
+            _id='testfile_nii',
+            name='testfile_nii',
+            materialized_path='/testfile_nii'
+        )
+        file_nii.save()
+
+        file_custom = OsfStorageFileNode.create(
+            target=self.node,
+            path='/testfile_custom',
+            _id='testfile_custom',
+            name='testfile_custom',
+            materialized_path='/testfile_custom'
+        )
+        file_custom.save()
+
+        # Create NII_STORAGE first
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 1000
+                }
+            },
+            file_node=file_nii,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        # Create CUSTOM_STORAGE
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 2000
+                }
+            },
+            file_node=file_custom,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+
+        nii_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        custom_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.CUSTOM_STORAGE)
+
+        assert_equal(nii_quota.used, 1000)
+        assert_equal(custom_quota.used, 2000)
+
+    def test_get_or_create_preserves_existing_max_quota(self):
+        """get_or_create should not overwrite existing max_quota when record exists"""
+        custom_max = 300
+        UserQuota.objects.create(
+            user=self.project_creator,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=custom_max,
+            used=0
+        )
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 1500
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.max_quota, custom_max)
+        assert_equal(user_quota.used, 1500)
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_get_or_create_without_select_for_update_new_userquota(self, _mock_check_select):
+        """get_or_create without select_for_update should create new UserQuota"""
+        assert_false(UserQuota.objects.filter(user=self.project_creator).exists())
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 5000
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(len(user_quota), 1)
+        user_quota = user_quota[0]
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 5000)
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_get_or_create_without_select_for_update_existing_userquota(self, _mock_check_select):
+        """get_or_create without select_for_update should retrieve existing UserQuota"""
+        UserQuota.objects.create(
+            user=self.project_creator,
+            storage_type=UserQuota.NII_STORAGE,
+            max_quota=200,
+            used=100
+        )
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 200
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota_list = UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(len(user_quota_list), 1)
+        user_quota = user_quota_list[0]
+        assert_equal(user_quota.max_quota, 200)
+        assert_equal(user_quota.used, 300)
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_get_or_create_without_select_for_update_custom_storage(self, mock_check_select):
+        """get_or_create without select_for_update should work with CUSTOM_STORAGE"""
+        institution = InstitutionFactory()
+        self.project_creator.affiliated_institutions.add(institution)
+        RegionFactory(_id=institution._id)
+        ProjectStorageType.objects.filter(node=self.node).update(
+            storage_type=ProjectStorageType.CUSTOM_STORAGE
+        )
+
+        assert_false(UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.CUSTOM_STORAGE).exists())
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 3000
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.CUSTOM_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.CUSTOM_STORAGE)
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+        assert_equal(user_quota.used, 3000)
+
+    @mock.patch('website.util.quota.check_select_for_update')
+    def test_file_modified_else_branch_fileinfo_get(self, mock_check):
+        """Test the else branch at line 330 for FileInfo.get without select_for_update"""
+        mock_check.return_value = False
+
+        # Create initial FileInfo
+        FileInfo.objects.create(file=self.file, file_size=500)
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 1200
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        # used = 1200 - 500 = 700 (retrieved existing FileInfo via else branch)
+        assert_equal(user_quota.used, 700)
+
+        # Verify check_select_for_update was called multiple times (for UserQuota and FileInfo)
+        assert_true(mock_check.called)
+
+    @mock.patch('website.util.quota.check_select_for_update', return_value=False)
+    def test_file_modified_without_select_for_update_fileinfo_not_exists(self, mock_check_select):
+        """file_modified should handle FileInfo.DoesNotExist without select_for_update"""
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 2000
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        # used = 2000 - 0 = 2000 (no FileInfo exists, so old size is 0)
+        assert_equal(user_quota.used, 2000)
+
+        file_info = FileInfo.objects.get(file=self.file)
+        assert_equal(file_info.file_size, 2000)
+
+    @mock.patch('website.util.quota.check_select_for_update')
+    def test_file_modified_else_branch_get_or_create(self, mock_check):
+        """Test the else branch at line 315 for get_or_create without select_for_update"""
+        mock_check.return_value = False
+
+        assert_false(UserQuota.objects.filter(user=self.project_creator, storage_type=UserQuota.NII_STORAGE).exists())
+
+        quota.file_modified(
+            target=self.node,
+            payload={
+                'metadata': {
+                    'size': 1500
+                }
+            },
+            file_node=self.file,
+            storage_type=UserQuota.NII_STORAGE
+        )
+
+        # Verify that the else branch (line 315) was executed successfully
+        user_quota = UserQuota.objects.get(user=self.project_creator, storage_type=UserQuota.NII_STORAGE)
+        assert_equal(user_quota.used, 1500)
+        assert_equal(user_quota.max_quota, api_settings.DEFAULT_MAX_QUOTA)
+
+        # Verify check_select_for_update was called
+        assert_true(mock_check.called)
diff --git a/website/util/quota.py b/website/util/quota.py
index 1990073a444..195dae660f8 100644
--- a/website/util/quota.py
+++ b/website/util/quota.py
@@ -13,6 +13,7 @@
     ProjectStorageType
 )
 from django.utils import timezone
+from osf.models.institution_default_max_quota import InstitutionDefaultMaxQuota
 from osf.utils.requests import check_select_for_update
 
 
@@ -72,6 +73,7 @@ def update_user_used_quota(user, storage_type=UserQuota.NII_STORAGE, is_recalcul
         # Get total file size of projects with specified storage_type
         used = used_quota(user._id, storage_type)
 
+    max_quota = get_default_max_quota(user, storage_type)
     try:
         if check_select_for_update():
             user_quota = UserQuota.objects.filter(
@@ -91,10 +93,11 @@ def update_user_used_quota(user, storage_type=UserQuota.NII_STORAGE, is_recalcul
                 UserQuota.objects.create(
                     user=user,
                     storage_type=storage_type,
-                    max_quota=api_settings.DEFAULT_MAX_QUOTA,
+                    max_quota=max_quota,
                     used=used,
                 )
-        except IntegrityError:
+        except IntegrityError as e:
+            logger.warning(u'IntegrityError while creating UserQuota: user={}, storage_type={}: {}'.format(user.id, storage_type, str(e)))
             if is_recalculating_quota and storage_type == UserQuota.CUSTOM_STORAGE:
                 used_quota_for_nii_default_storage = used_quota(user._id, UserQuota.NII_STORAGE)
                 used_quota_for_nii_custom_storage = used_quota(user._id, UserQuota.CUSTOM_STORAGE)
@@ -128,11 +131,12 @@ def abbreviate_size(size):
     return (size, abbr_dict[power])
 
 def get_quota_info(user, storage_type=UserQuota.NII_STORAGE):
+    max_quota = get_default_max_quota(user, storage_type)
     try:
         user_quota = user.userquota_set.get(storage_type=storage_type)
         return (user_quota.max_quota, user_quota.used)
     except UserQuota.DoesNotExist:
-        return (api_settings.DEFAULT_MAX_QUOTA, used_quota(user._id, storage_type))
+        return (max_quota, used_quota(user._id, storage_type))
 
 def get_project_storage_type(node):
     try:
@@ -214,7 +218,7 @@ def update_used_quota(self, target, user, event_type, payload):
             else:
                 node_removed(target, user, payload, file_node, storage_type)
         elif event_type == FileLog.FILE_UPDATED:
-            file_modified(target, user, payload, file_node, storage_type)
+            file_modified(target, payload, file_node, storage_type)
     else:
         return
 
@@ -223,6 +227,7 @@ def file_added(target, payload, file_node, storage_type):
     file_size = int(payload['metadata']['size'])
     if file_size < 0:
         return
+    max_quota = get_default_max_quota(target.creator, storage_type)
     try:
         if check_select_for_update():
             user_quota = UserQuota.objects.filter(
@@ -242,10 +247,11 @@ def file_added(target, payload, file_node, storage_type):
                 UserQuota.objects.create(
                     user=target.creator,
                     storage_type=storage_type,
-                    max_quota=api_settings.DEFAULT_MAX_QUOTA,
+                    max_quota=max_quota,
                     used=file_size
                 )
-        except IntegrityError:
+        except IntegrityError as e:
+            logger.warning(u'IntegrityError while creating UserQuota in file_added: user={}, storage_type={}: {}'.format(target.creator.id, storage_type, str(e)))
             if check_select_for_update():
                 user_quota = UserQuota.objects.filter(
                     user=target.creator,
@@ -294,26 +300,27 @@ def node_removed(target, user, payload, file_node, storage_type):
             file_info.save()
         user_quota.save()
 
-def file_modified(target, user, payload, file_node, storage_type):
+def file_modified(target, payload, file_node, storage_type):
     file_size = int(payload['metadata']['size'])
     if file_size < 0:
         return
-
+    max_quota = get_default_max_quota(target.creator, storage_type)
     try:
         with transaction.atomic():
             if check_select_for_update():
                 user_quota, _ = UserQuota.objects.select_for_update().get_or_create(
                     user=target.creator,
                     storage_type=storage_type,
-                    defaults={'max_quota': api_settings.DEFAULT_MAX_QUOTA}
+                    defaults={'max_quota': max_quota}
                 )
             else:
                 user_quota, _ = UserQuota.objects.get_or_create(
                     user=target.creator,
                     storage_type=storage_type,
-                    defaults={'max_quota': api_settings.DEFAULT_MAX_QUOTA}
+                    defaults={'max_quota': max_quota}
                 )
-    except IntegrityError:
+    except IntegrityError as e:
+        logger.warning(u'IntegrityError while creating UserQuota in file_modified: user={}, storage_type={}: {}'.format(target.creator.id, storage_type, str(e)))
         if check_select_for_update():
             user_quota = UserQuota.objects.filter(user=target.creator, storage_type=storage_type).select_for_update().get()
         else:
@@ -390,3 +397,31 @@ def get_node_file_list(file_node):
                 file_list.append(child_file_node)
 
     return file_list
+
+def get_default_max_quota(user, storage_type):
+    """
+    Get default max quota for a user based on their affiliated institution.
+
+    If the storage type is CUSTOM_STORAGE, retrieve the default quota
+    configured for the user's affiliated institution. Otherwise, or if no
+    institution-specific quota is configured, return the system default quota.
+    When the institution's region uses NII_STORAGE, always return the system default quota.
+
+    Args:
+        user (OSFUser): The user whose default quota is to be retrieved.
+        storage_type (str): The storage type associated with the user's quota.
+
+    Returns:
+        int: The default maximum quota (in GB) applicable to the user.
+    """
+    if storage_type == UserQuota.CUSTOM_STORAGE:
+        institution = user.representative_affiliated_institution
+        if institution and not Region.objects.filter(
+            _id=institution._id,
+            waterbutler_settings__storage__type=Region.NII_STORAGE
+        ).exists():
+            default_max_quota = InstitutionDefaultMaxQuota.get_quota_by_user(user.id)
+            if default_max_quota is not None:
+                return default_max_quota
+
+    return api_settings.DEFAULT_MAX_QUOTA